Usage Guide
Registering Dependencies
To register a dependency pass provider you want into Container.register
:
from aioinject import Container, Scoped
class MyClass:
pass
container = Container()
container.register(Scoped(MyClass))
Generic Dependencies
Aioinject supports registering unbound generic classes and passing generic parameters when resolving them.
from typing import Final, Generic, TypeVar
from aioinject import Object, Scoped, SyncContainer
T = TypeVar("T")
class Box(Generic[T]):
def __init__(self, value: T) -> None:
self.value: Final = value
def __repr__(self) -> str:
return f"Box({self.value!r})"
container = SyncContainer()
container.register(
Scoped(Box),
Object("string value"),
Object(42),
)
with container, container.context() as context:
int_box = context.resolve(Box[int])
print(int_box) # Box(42)
str_box = context.resolve(Box[str])
print(str_box) # Box('string value')
container.register(Object(Box("bound"), interface=Box[str]))
box_str = context.resolve(Box[str])
print(box_str) # Box('bound')
container.register(Object(Box("bound"), interface=Box[str]))
with container.context() as context:
box_str = context.resolve(Box[str])
print(box_str) # Box('bound')
Iterable dependencies
Sometimes there's a need to register and resolve multiple dependencies of the same type/interface.
Iterable dependencies in aioinject work similarly to Enumerable
dependencies in C#
/.NET
- all dependencies
are instantiated and provided.
from collections.abc import Sequence
from aioinject import Singleton, SyncContainer
class Logger:
pass
class FileLogger(Logger):
pass
class DatabaseLogger(Logger):
pass
class StreamLogger(Logger):
pass
container = SyncContainer()
container.register(
Singleton(FileLogger, Logger),
Singleton(DatabaseLogger, Logger),
Singleton(StreamLogger, Logger),
)
with container, container.context() as context:
loggers = context.resolve(Sequence[Logger]) # type: ignore[type-abstract]
print(loggers) # [<FileLogger>, <DatabaseLogger>, <StreamLogger>]
Warning
When multiple providers are registered with same interface the most recent one would be provided:
Note
Currently iterable dependencies are always provided in a list
container.
Context Managers / Resources
Applications often need to close dependencies after they're done using them,
this can be done by registering a function decorated with @contextlib.contextmanager
or @contextlib.asynccontextmanager
.
Internally aioinject would use contextlib.ExitStack
or contextlib.AsyncExitStack
to manage them.
import contextlib
from collections.abc import Iterator
import aioinject
from aioinject import Scoped, Singleton
@contextlib.contextmanager
def singleton_dependency() -> Iterator[int]:
print("Singleton Startup")
yield 42
print("Singleton Shutdown")
@contextlib.contextmanager
def scoped_dependency(number: int) -> Iterator[str]:
print("Scoped Startup")
yield str(number)
print("Scoped Shutdown")
container = aioinject.SyncContainer()
container.register(Singleton(singleton_dependency))
container.register(Scoped(scoped_dependency))
with container: # noqa: SIM117
with container.context() as ctx:
value = ctx.resolve(str) # Singleton Startup, Scoped Startup
print(repr(value)) # '42'
# Scoped Shutdown
# Singleton Shutdown
Handling Errors
If Exceptions are raised inside a scope they're propagated into context managers, if you're not wrapping an
already existing context manager (e.g. SQLAlchemy's Session.begin
) you should use try-except-finally
to correctly close your dependencies.
@contextlib.contextmanager
def dependency() -> Iterator[int]:
obj = SomeObject()
try:
yield obj
except:
... # Error handling code
finally:
obj.close()
Managing Application Lifetime
In order for container to close singleton dependencies on application shutdown you need to use container as a context manager.
import asyncio
from aioinject import Container
async def main() -> None:
container = Container()
async with container:
...
if __name__ == "__main__":
asyncio.run(main())
LifespanExtension
and LifespanSyncExtension