modern-di-aiohttp 4.2.4

modern-di integration for aiohttp.

Full guide: aiohttp integration docs

Usage example: examples/

Installation

uv add modern-di-aiohttp      # or: pip install modern-di-aiohttp

Usage

aiohttp has no dependency-injection system of its own, so modern-di-aiohttp pairs an @inject decorator with inert FromDI markers. setup_di stores the container on the app, opens it on startup and closes it on cleanup, and installs middleware that builds a per-connection child container automatically; FromDI resolves a provider (or type) into a handler parameter.

import dataclasses
import typing

from aiohttp import web
from modern_di import Container, Group, Scope, providers
from modern_di_aiohttp import FromDI, inject, setup_di


@dataclasses.dataclass(kw_only=True)
class Settings:
    debug: bool = True


@dataclasses.dataclass(kw_only=True)
class UserService:
    settings: Settings  # auto-injected by type


class Dependencies(Group):
    settings = providers.Factory(scope=Scope.APP, creator=Settings)
    user_service = providers.Factory(scope=Scope.REQUEST, creator=UserService)


@inject
async def index(
    request: web.Request,
    user_service: typing.Annotated[UserService, FromDI(Dependencies.user_service)],
) -> web.Response:
    return web.json_response({"debug": user_service.settings.debug})


app = web.Application()
app.router.add_get("/", index)
container = Container(groups=[Dependencies])
setup_di(app, container)
container.validate()  # optional fail-fast; must come after setup_di registers its providers

@inject also works on a plain def handler. It returns a sync wrapper for a sync function, so aiohttp handles the decorated handler the same way it handles the undecorated one, including its deprecation warning for sync handlers.

@inject works the same on the methods of a class-based web.View. Decorate the handler method, not the class; aiohttp calls the method with no arguments, so the request is taken from self.request:

class Users(web.View):
    @inject
    async def get(
        self,
        user_service: typing.Annotated[UserService, FromDI(Dependencies.user_service)],
    ) -> web.Response:
        return web.json_response({"debug": user_service.settings.debug})


app.router.add_view("/users", Users)

An HTTP request opens a Scope.REQUEST child container; a WebSocket connection opens a Scope.SESSION one. The connection aiohttp.web.Request is resolvable within DI: HTTP handlers and REQUEST-scoped factories inject it by type via aiohttp_request_provider, while WebSocket handlers read it via FromDI(aiohttp_websocket_provider). For per-message work inside a WebSocket handler, open a nested Scope.REQUEST child of the session container fetched with fetch_request_container.

Sub-apps added with app.add_subapp work with or without their own setup_di. When both the parent and the sub-app call setup_di, a sub-app route resolves from a child of the sub-app's root container, and the parent's middleware still opens and closes its own child container around it. When only the parent calls setup_di, a sub-app route resolves from a child of the parent's root container.

Sharing a container

By default setup_di opens the root container on startup and closes it on cleanup. If the same container also serves an app of another modern-di integration, such as a FastStream consumer, the first app to shut down would close it for the rest. Pass manage_lifespan=False to every setup_di that shares the container and own the root yourself. The middleware still builds and closes a child container per connection. See FastAPI and FastStream sharing a container for the pattern.

import asyncio


async def main() -> None:
    app = web.Application()
    container = Container(groups=[Dependencies])
    setup_di(app, container, manage_lifespan=False)

    async with container:
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "localhost", 8080)
        await site.start()
        try:
            await asyncio.Event().wait()
        finally:
            await runner.cleanup()

APP-scoped finalizers run once, when the async with block exits.

Shutdown order

setup_di appends the root container's close to app.on_cleanup. These hooks run before the close and can still resolve from the container:

• on_shutdown handlers
• cleanup_ctx teardown
• on_cleanup handlers added before setup_di

on_cleanup handlers added after setup_di run after the close and get ContainerClosedError. Add them before setup_di. If you can't, pass manage_lifespan=False and close the container yourself after runner.cleanup(), as in the example above.

API

Links

• PyPI
• License

Part of modern-python

Built on modern-di, a dependency-injection framework with an IoC container and scopes.

Browse the full list of templates and libraries in modern-python; the org profile has the categorized index.