AIOHTTP Integration

Dependency injection for AIOHTTP is available in the wireup.integration.aiohttp module.

• Automatic Dependency Management

  ---

  Inject dependencies in routes and automatically manage container lifecycle.

• Request Objects

  ---

  Use request and websocket objects in Wireup dependencies.

• Zero Runtime Overhead

  ---

  Inject dependencies with zero runtime overhead in Class-Based Handlers.

  Learn more

• Shared business logic

  ---

  Wireup is framework-agnostic. Share the service layer between web applications and other interfaces, such as a CLI.

Initialize the integration

First, create an async container.

container = wireup.create_async_container(
    injectables=[services], config={"db_dsn": os.environ.get("APP_DB_DSN")}
)

Then initialize the integration:

wireup.integration.aiohttp.setup(container, app)

Inject in AIOHTTP handlers

To inject dependencies, add the type to the handler's signature and annotate with Injected[T] or Annotated[T, Inject(...)].

Function HandlersClass-Based Views

Function Handler

async def get_users(
    request: web.Request,
    user_repository: Injected[UserRepository],
) -> web.Response: ...

In Class-based views dependencies must be declared in the init method.

Class Based View

class UsersView(web.View):
    def __init__(
        self,
        request: web.Request,
        user_repository: Injected[UserRepository],
    ) -> None:
        super().__init__(request)
        self.user_repository = user_repository

    async def get(self) -> web.Response: ...

Inject AIOHTTP request

To be able to inject web.Request, include the wireup.integration.aiohttp module in the injectables parameter when creating the container.

container = wireup.create_async_container(
    injectables=[services, wireup.integration.aiohttp],
    config={"db_dsn": os.environ.get("APP_DB_DSN")},
)

Accessing the Container

If you need to access the Wireup container directly, use the following functions:

from wireup.integration.aiohttp import get_app_container, get_request_container

# Get application-wide container.
app_container: AsyncContainer = get_app_container(app)

# Get request-scoped container.
# This is what is currently injecting services on the active request.
request_container: ScopedAsyncContainer = get_request_container()

Container Availability and middleware_mode

get_app_container(app) is always available.

get_request_container() is available only when middleware_mode=True. In AIOHTTP integration this is the default behavior:

wireup.integration.aiohttp.setup(
    container, app, middleware_mode=True
)  # default

This mode adds AIOHTTP middleware so the request-scoped container is available everywhere in request lifecycle code (for example decorators and custom middleware), not just in Wireup-injected handlers. All normal Wireup handler injection keeps working either way.

If you do not need get_request_container(), disable it for lower request-path overhead:

wireup.integration.aiohttp.setup(container, app, middleware_mode=False)

With middleware_mode=False, Wireup still injects dependencies in handlers, including web.Request when wireup.integration.aiohttp is included in injectables, but get_request_container() is not available.

Use middleware_mode=True when you need to access the request-scoped Wireup container from places where Wireup is not injecting directly (for example AIOHTTP middleware or custom decorator paths).

Testing

For general testing tips with Wireup refer to the test docs. With the AIOHTTP integration, you can override dependencies in the container as follows.

test_thing.py

from wireup.integration.aiohttp import get_app_container


def test_override(aiohttp_client):
    class DummyGreeter(GreeterService):
        def greet(self, name: str) -> str:
            return f"Hi, {name}"

    with get_app_container(app).override.injectable(
        GreeterService,
        new=DummyGreeter(),
    ):
        res = aiohttp_client.get("/greet?name=Test")

See AIOHTTP integration tests for more examples.

Lifecycle Management

The integration automatically manages the container lifecycle. It hooks into AIOHTTP's on_cleanup signal to ensure the container is properly closed and resources are released when the application stops.

Routes and type checker

If you're using a type checker, then you may notice it showing type errors when adding dependencies to aio handlers. This is because the signature as defined in aiohttp only allows for web.Request in the signature.

To address this, you can decorate the handler with @wireup.integration.aiohttp.route. This decorator is a no-op at runtime but provides the necessary type hints to satisfy type checkers.

from wireup.integration.aiohttp import route


@router.get("/users")
@route
async def users_list(
    request: web.Request,
    user_repository: Injected[UserRepository],
) -> web.Response:
    pass

API Reference

• aiohttp_integration