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()
andaget()
in other middlewares.
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=False, ping=None, on_registry_close=None)[source]¶
Same as
svcs.Registry.register_value()
, but uses registry on app.
Service Acquisition¶
- 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