AIOHTTP

svcs’s AIOHTTP integration stores the svcs.Registry on the aiohttp.web.Application object and the svcs.Container on the aiohttp.web.Request object.

Initialization

You add support for svcs to your AIOHTTP app by calling svcs.aiohttp.init_app() wherever you create your aiohttp.web.Application object.

Service Acquisition

You can get either the Container using svcs.aiohttp.svcs_from() and use it as usual, or pluck them directly from the request object using svcs.aiohttp.aget() that takes a aiohttp.web.Request object as its first argument.

Health Checks

As with services, you have the option to either svcs.aiohttp.svcs_from() on the request or go straight for svcs.aiohttp.get_pings().

A health endpoint could look like this:

from __future__ import annotations

from aiohttp.web import Request, Response, json_response

import svcs


async def healthy_view(request: Request) -> Response:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: dict[str, str] = {}
    code = 200

    for svc in svcs.aiohttp.get_pings(request):
        try:
            await svc.aping()
            ok.append(svc.name)
        except Exception as e:
            failing[svc.name] = repr(e)
            code = 500

    return json_response({"ok": ok, "failing": failing}, status=code)

Cleanup

Acquired services and pings get cleaned up automatically at the end of a request.

If you register on_registry_close callbacks, you can use svcs.aiohttp.aclose_registry() to run them. init_app() will automatically add them to the app’s aiohttp.web.Application.on_cleanup callbacks. Therefore, if you shut down your AIOHTTP applications cleanly, you don’t have to think about registry cleanup either.

API Reference

Application Life Cycle

svcs.aiohttp.init_app(app, *, registry=None, middleware_pos=0)

Initialize the application.

Inserts the svcs middleware at middleware_pos which is 0 by default, so you can use svcs_from() and aget() in other middlewares.

async svcs.aiohttp.aclose_registry(app)

Close the registry on app, if present.

You probably don’t have to call this yourself, because it’s registered for the application as an {attr}`aiohttp.web.Application.on_cleanup` callback.

See also

Cleanup

svcs.aiohttp.get_registry(app)

Get the registry from app.

Registering Services

svcs.aiohttp.register_factory(app, svc_type, factory, *, enter=True, ping=None, on_registry_close=None)

Same as svcs.Registry.register_factory(), but uses registry on app.

svcs.aiohttp.register_value(app, svc_type, value, *, enter=False, ping=None, on_registry_close=None)

Same as svcs.Registry.register_value(), but uses registry on app.

Service Acquisition

svcs.aiohttp.svcs_from(request)

Get the current container from request.

async svcs.aiohttp.aget(request: aiohttp.web.Request, svc_type1: type, ...)

Same as svcs.Container.aget(), but uses the container from request.

async svcs.aiohttp.aget_abstract(request, *svc_types)

Same as svcs.Container.aget_abstract(), but uses container from request.

svcs.aiohttp.get_pings(request)

Same as svcs.Container.get_pings(), but uses the container from request.

See also

Health Checks