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(
    service_modules=[services],
    parameters={"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 them as necessary. See Annotations for more details.

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() -> web.Response: ...

Inject AIOHTTP request

To inject web.Request in services, include wireup.integration.aiohttp module in the service modules when creating the container.

container = wireup.create_async_container(
    service_modules=[
        services,
        wireup.integration.aiohttp
    ],
    parameters={"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.service(GreeterService, new=DummyGreeter()):
        res = aiohttp_client.get("/greet?name=Test")

See AIOHTTP integration tests for more examples.

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 parameters. To make the type checker happy you can annotate them with wireup.integration.aiohttp.route.

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