Skip to content

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 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()

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