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: list[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.append({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)[source]#

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)[source]#

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)[source]#

Get the registry from app.

Registering Services#

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

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

svcs.aiohttp.register_value(app, svc_type, value, *, enter=True, ping=None, on_registry_close=None)[source]#

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

Service Acquisition#

svcs.aiohttp.svcs_from(request)[source]#

Get the current container from request.

async svcs.aiohttp.aget(request: aiohttp.web.Request, svc_type1: type, ...)[source]#

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

async svcs.aiohttp.aget_abstract(request, *svc_types)[source]#

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

svcs.aiohttp.get_pings(request)[source]#

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

See also

Health Checks