Injex

Async

Async resolution

The synchronous API stays exactly as it is. Async support is a separate path you opt into when a dependency is genuinely asynchronous — an async def factory, or a resource that must be opened and closed around its lifetime. It uses the standard library's contextlib.AsyncExitStack, so still no runtime dependencies.

Async factories

Register a coroutine-function factory and resolve it through the async API. The result is cached per its lifetime just like a sync factory: 

async def load_settings() -> Settings:
    return await read_config()

container.add_singleton_factory(Settings, load_settings)

async def main() -> None:
    settings = await container.aresolve(Settings)

The sync resolve() raises AsyncResolutionRequiredException if the graph needs async work, so you never silently get back an un-awaited coroutine: 

container.resolve(Settings)  # AsyncResolutionRequiredException

Async "infects" upward: a plain synchronous class that depends on an async factory is also resolved through the async path.

Async resources

A resource has a lifetime: a database session, a connection pool, an HTTP client. Declare it as an async generator factory: 

from typing import AsyncIterator

async def db_session(pool: Pool) -> AsyncIterator[Session]:
    session = await pool.acquire()
    try:
        yield session
    finally:
        await session.close()

container.add_scoped_factory(Session, db_session)

When the resource is finalized depends on its lifetime:

Finalization is LIFO, like nested async with blocks.

Async scopes

Open a scope with async with. Resources resolved inside are closed when the block exits: 

async def handle_request() -> None:
    async with container.ascope() as scope:
        service = await scope.aresolve(RegisterUser)
        await service.execute(...)
    # scoped resources finalized here, LIFO

aresolve() on the container (outside a scope) refuses to open a scoped/transient resource it would have to finalize immediately; resolve those inside async with container.ascope() instead.

FastAPI

Build the container once in lifespan, open one async scope per request, and close singleton resources on shutdown: 

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.container = build_container()
    yield
    await app.state.container.aclose()

async def request_scope(request: Request) -> AsyncIterator[AsyncScope]:
    async with request.app.state.container.ascope() as scope:
        yield scope

async def get_service(scope: AsyncScope = Depends(request_scope)) -> RegisterUser:
    return await scope.aresolve(RegisterUser)

Full runnable version: examples/fastapi_async.py.

Notes