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 your web application and other interfaces, such as a CLI.

Initialize the integration

To initialize the integration, call wireup.integration.aiohttp.setup.

container = wireup.create_async_container(
    # Add service modules.
    service_modules=[
        # Top level module containing service registrations.
        services,
        # Add this module if you need `web.Request` in Wireup services.
        wireup.integration.aiohttp
    ],
    # Expose parameters to Wireup as necessary. 
    parameters={"db_dsn": os.environ.get("APP_DB_DSN")}
)
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

A key feature of the integration is to expose web.Request in Wireup. To inject it in your services you must add wireup.integration.aiohttp module to your service modules when creating a container.

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