Injection¶
Injection is the main purpose of this library. This is the process of making it all work and sound.
- Injection can be
Synchronous — performed by
injectfunction. Supports only synchronous dependencies and dependants. If any dependency or dependant is asynchronous — useainjectinstead.Asynchronous — performed by
ainjectfunction. Supports both async and sync dependencies and dependants. FunDI handles both sync and async dependencies transparently, even if they’re mixed.Note that synchronous dependencies would not be called in a thread. So it is safe to do event-loop related stuff in synchronous dependencies.
Example of synchronous injection:
import secrets
from contextlib import ExitStack
from fundi import Scope, from_, scan, inject
def require_unique_id() -> str:
return secrets.token_hex(12)
def application(username: str, user_id: str = from_(require_unique_id)) -> None:
print(f"Application started with {user_id = } and {username = }")
inject(Scope({"username": "Kuyugama"}), scan(application))
Example of asynchronous injection:
import secrets
import asyncio
from contextlib import AsyncExitStack
from fundi import Scope, from_, scan, ainject
async def require_user(username: str) -> str:
await asyncio.sleep(0.4) # pretend to be making web request
return {"id": secrets.token_hex(12), "username": username}
def application(user: dict[str, str] = from_(require_user)) -> None:
print(f"Application started with {user['id'] = } and {user['username'] = }")
async def main():
await ainject(Scope({"username": "Kuyugama"}), scan(application))
if __name__ == "__main__":
asyncio.run(main())
Example of injection that produces value:
import secrets
from contextlib import ExitStack
from fundi import Scope, scan, inject
def require_unique_id() -> str:
return secrets.token_hex(12)
user_id = inject(Scope(), scan(require_unique_id))
print("Generated user id is", user_id)
The same works with asynchronous injection
Own exit stack¶
You may want to control when dependencies are torn down. To do so, provide your own exit stack. When you need dependencies to be torn down, simply close the stack — that’s it!
Synchronous example¶
import time
from contextlib import ExitStack
from fundi import inject, scan, from_, Scope
def dependency(value: str = "default"):
print("set-up")
yield value
print("tear-down")
def dependant(value: str = from_(dependency)):
yield "Dependant got " + value
with ExitStack() as stack:
value = inject(Scope({"value": "value"}), scan(dependant), stack)
print(value)
print("Doing some computations before dependencies would tear-down")
time.sleep(0.8) # simulate computations
Asynchronous example¶
import asyncio
from contextlib import AsyncExitStack
from fundi import inject, scan, from_, Scope
def dependency(value: str = "default"):
print("set-up")
yield value
print("tear-down")
def dependant(value: str = from_(dependency)):
yield "Dependant got " + value
async def main():
async with AsyncExitStack() as stack:
value = inject(Scope({"value": "value"}), scan(dependant), stack)
print(value)
print("Doing some computations before dependencies would tear-down")
await asyncio.sleep(0.8) # simulate computations
if __name__ == "__main__":
asyncio.run(main())
In earlier versions, FunDI required an explicit ExitStack or AsyncExitStack to be provided for every injection.
It is now optional — if not provided, FunDI will create and manage one automatically.
Dependency parameter awareness¶
Sometimes dependencies need to know where they are being injected.
With dependency parameter awareness, FunDI makes the fundi.Parameter object of the target parameter available for any dependency that declares:
from fundi import FromType, Parameter
def dependency(param: FromType[Parameter]):
print(param.name) # name of the parameter being injected to
print(param.annotation) # expected type of the parameter
...
Note: Parameter-aware dependencies (i.e., those that accept a
FromType[Parameter]) are cached just like any other dependency by default.This means that even if the same function is injected into multiple parameters (e.g.,
user_id,client_id), it will only be called once, and the cached result will be reused — regardless of which parameter it’s injected into.If your function depends on the parameter name or annotation (e.g. to extract different headers), you must disable caching manually using
from_(..., caching=False).This behavior is intentional for now and may change in future versions, but currently it’s the developer’s responsibility to manage it.
This allows you to build smarter and more reusable dependencies, such as:
Automatically inferring names (e.g.
user_id: int = from_header()→X-User-Idfrom parameter name)Performing type conversion or validation based on annotation
Example¶
from fundi import scan, FromType, from_, inject, Parameter, Scope def header(param: FromType[Parameter]) -> str: print("Parameter name:", param.name) print("Parameter annotation:", param.annotation) assert param.name == "token" assert param.annotation is str return f"{param.name}-{param.annotation!r}" def application(token: str = from_(header)): print("Token:", token) assert token == f"token-{str!r}" inject(Scope(), scan(application))
Summary¶
Feature |
|
|
|---|---|---|
Sync dependencies only |
Yes |
Yes |
Async dependencies |
No |
Yes |
Mixed (sync + async) |
No |
Yes |
Exit stack |
|
|
Dependency parameter awareness |
Yes |
Yes |
Returns value |
Yes |
Yes |