Core Concepts#

To understand how svcs works regardless of your environment, you must understand only two concepts: registries and containers. They have different life cycles and different responsibilities.

I practice, you will use one of our framework integrations (or write your own) and not the low-level API directly – but knowing what’s happening underneath is good to dispel any concerns about magic.

Registries#

A svcs.Registry allows registering factories for types. A registry should live as long as your application lives and there should be only one per application. Its only job is to store and retrieve factories along with some metadata.

It is possible to register either factory callables or values:

>>> import svcs
>>> import uuid

>>> reg = svcs.Registry()

>>> reg.register_factory(uuid.UUID, uuid.uuid4)
>>> reg.register_value(str, "Hello World")
>>> uuid.UUID in reg
True
>>> str in reg
True
>>> int in reg
False

The values and return values of the factories don’t have to be actual instances of the type they’re registered for. But the types must be hashable because they’re used as keys in a lookup dictionary.

Cleanup#

It’s possible to register a callback that is called when the registry is closed:

registry.register_factory(
    Connection, connection_factory, on_registry_close=engine.dispose
)

If this callback fails, it’s logged at warning level but otherwise ignored. For instance, you could free a database connection pool in an atexit handler or pytest fixture. This liberates you from keeping track of registered services yourself. You can also use a registry as an (async) context manager that (a)closes automatically on exit.

svcs will raise a ResourceWarning if a registry with pending cleanups is garbage-collected.

Containers#

A svcs.Container uses a svcs.Registry to lookup registered types and uses that information to create instances and to take care of their life cycles:

>>> container = svcs.Container(reg)

>>> uuid.UUID in container
False
>>> u = container.get(uuid.UUID)
>>> u
UUID('...')
>>> uuid.UUID in container
True
>>> # Calling get() again returns the SAME UUID instance!
>>> # Good for DB connections, bad for UUIDs.
>>> u is container.get(uuid.UUID)
True
>>> container.get(str)
'Hello World'

A container lives as long as you want the instances within to live – for example, as long as a request lives.

If a factory takes a first argument called svcs_container or the first argument (of any name) is annotated as being svcs.Container, the current container instance is passed into the factory as the first positional argument allowing for recursive service acquisition:

>>> container = svcs.Container(reg)

# Let's make the UUID predictable for our test!
>>> reg.register_value(uuid.UUID, uuid.UUID('639c0a5c-8d93-4a67-8341-fe43367308a5'))

>>> def factory(svcs_container) -> str:
...     return svcs_container.get(uuid.UUID).hex  # get the UUID, then work on it

>>> reg.register_factory(str, factory)

>>> container.get(str)
'639c0a5c8d934a678341fe43367308a5'

Note

It is possible to overwrite registered service factories later – for example, for testing – without monkey-patching. This is especially interesting if you want to replace a low-level service with a mock without re-jiggering all services that depend on it.

If there’s a chance that the container has been used by your fixtures to acquire a service, it’s possible that the service is already cached by the container. In this case make sure to reset it by calling svcs.Container.close() on it after overwriting. Closing a container is idempotent and it’s safe to use it again afterwards.

If your integration has a function called overwrite_(value|factory)(), it will do all of that for you.

Cleanup#

If a factory returns a context manager, it will be immediately entered and the instance will be added to the cleanup list (you can disabled this behavior by passing enter=False to register_factory() and register_value()). If a factory is a generator that yields the instance instead of returning it, it will be wrapped in a context manager automatically. At the end, you run svcs.Container.close() and all context managers will be exited. You can use this to close files, return database connections to a pool, and so on.

Async context managers and async generators work the same way.

You can also use containers as (async) context managers that (a)close automatically on exit:

>>> reg = svcs.Registry()
>>> def factory() -> str:
...     yield "Hello World"
...     print("Cleaned up!")
>>> reg.register_factory(str, factory)

>>> with svcs.Container(reg) as con:
...     _ = con.get(str)
Cleaned up!

Failing cleanups are logged at warning level but otherwise ignored.

Important

The key idea is that your business code doesn’t have to care about cleaning up services it has acquired.

That makes testing even easier because the business code makes fewer assumptions about the object it’s getting.

svcs will raise a ResourceWarning if a container with pending cleanups is garbage-collected.

Health Checks#

Each registered service may have a ping callable that you can use for health checks. You can request all pingable registered services with svcs.Container.get_pings(). This returns a list of svcs.ServicePing objects that currently have a name property to identify the ping and a ping() method that instantiates the service, adds it to the cleanup list, and runs the ping.

If you have async services (factory or ping callable), you can use aping() instead. aping() works with sync services, too, so you can use it universally in async code. You can look at the is_async property to check whether you need to use aping(), though.

Here’s how a health check endpoint could look like:

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)
from __future__ import annotations

from fastapi import FastAPI
from fastapi.responses import JSONResponse

import svcs


app = FastAPI(...)


@app.get("/healthy")
async def healthy(services: svcs.fastapi.DepContainer) -> JSONResponse:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: list[dict[str, str]] = []
    code = 200

    for svc in services.get_pings():
        try:
            await svc.aping()
            ok.append(svc.name)
        except Exception as e:
            failing.append({svc.name: repr(e)})
            code = 500

    return JSONResponse(
        content={"ok": ok, "failing": failing}, status_code=code
    )
from __future__ import annotations

import flask

import svcs


bp = flask.Blueprint("instrumentation", __name__)


@bp.get("healthy")
def healthy() -> flask.ResponseValue:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: list[dict[str, str]] = []
    code = 200

    for svc in svcs.flask.get_pings():
        try:
            svc.ping()
            ok.append(svc.name)
        except Exception as e:
            failing.append({svc.name: repr(e)})
            code = 500

    return {"ok": ok, "failing": failing}, code
from __future__ import annotations

import json

from pyramid.request import Request
from pyramid.response import Response
from pyramid.view import view_config

import svcs


@view_config(route_name="healthy")
def healthy_view(request: Request) -> Response:
    """
    Ping all external services.
    """
    ok: list[str] = []
    failing: list[dict[str, str]] = []
    status = 200

    for svc in svcs.pyramid.get_pings(request):
        try:
            svc.ping()
            ok.append(svc.name)
        except Exception as e:
            failing.append({svc.name: repr(e)})
            status = 500

    return Response(
        content_type="application/json",
        status=status,
        body=json.dumps({"ok": ok, "failing": failing}).encode("ascii"),
    )
from __future__ import annotations

from starlette.requests import Request
from starlette.responses import JSONResponse

import svcs


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

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

    return JSONResponse(
        content={"ok": ok, "failing": failing}, status_code=code
    )

Now, you can point your monitoring tool of choice – like Prometheus’s Blackbox Exporter or Nagios – at it and you’ll get alerted whenever the application is broken.

Life Cycle Summary#

While svcs’s core is entirely agnostic on how you use the registry and the container, all our Integrations follow the same life cycle:

You’re free to structure your own integrations as you want, though.

Debugging Registrations#

If you are confused about where a particular factory for a type has been defined, svcs logs every registration at debug level along with a stack trace.

Set the svcs logger to DEBUG to see them:

import logging.config

from datetime import datetime

import svcs


logging.config.dictConfig(
    {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "standard": {
                "format": "%(name)s: %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "level": "DEBUG",
                "formatter": "standard",
            },
        },
        "loggers": {
            "svcs": {
                "handlers": ["console"],
                "level": "DEBUG",
                "propagate": False,
            },
        },
        "root": {
            "handlers": ["console"],
            "level": "INFO",
        },
    }
)


reg = svcs.Registry()

reg.register_factory(datetime, datetime.now)
reg.register_value(str, "Hello World")

It gives you an output like this:

svcs: registered factory <built-in method now of type object at 0x103468980> for service type datetime.datetime
Stack (most recent call last):
  File "/Users/hynek/FOSS/svcs/docs/examples/debugging_with_logging.py", line 41, in <module>
    reg.register_factory(datetime, datetime.now)
  File "/Users/hynek/FOSS/svcs/src/svcs/_core.py", line 216, in register_factory
    log.debug(
svcs: registered value 'Hello World' for service type builtins.str
Stack (most recent call last):
  File "/Users/hynek/FOSS/svcs/docs/examples/debugging_with_logging.py", line 42, in <module>
    reg.register_value(str, "Hello World")
  File "/Users/hynek/FOSS/svcs/src/svcs/_core.py", line 252, in register_value
    log.debug(

You can see that the datetime factory and the str value have both been registered in debugging_with_logging.py, down to the line number.

API Reference#

class svcs.Registry#

A central registry of recipes for creating services.

An instance of this should live as long as your application does.

Also works as a context manager that runs on_registry_close callbacks on exit:

>>> import svcs
>>> with svcs.Registry() as reg:
...     reg.register_value(
...         int, 42,
...         on_registry_close=lambda: print("closed!")
...     )
closed!

async with is also supported.

Warns:

ResourceWarning – If a registry with pending cleanups is garbage-collected.

__contains__(svc_type)[source]#

Check whether this registry knows how to create svc_type:

>>> reg = svcs.Registry()
>>> reg.register_value(int, 42)
>>> int in reg
True
>>> str in reg
False
async aclose()[source]#

Clear registrations and run all on_registry_close callbacks.

Errors are logged at warning level, but otherwise ignored.

Also works with synchronous services, so in an async application, just use this.

close()[source]#

Clear registrations and run synchronous on_registry_close callbacks.

Async callbacks are not awaited and a warning is raised

Errors are logged at warning level, but otherwise ignored.

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

Register factory to be used when asked for a svc_type.

Repeated registrations overwrite previous ones, but the on_registry_close callbacks are run all together when the registry is closed.

Parameters:
  • svc_type (type) – The type of the service to register.

  • factory (Callable) –

    A callable that is used to instantiated svc_type if asked. If it’s a generator or a context manager, a cleanup is registered after instantiation.

    Can also be an async callable/generator/context manager..

    If factory takes a first argument called svcs_container or the first argument (of any name) is annotated as being svcs.Container, the container instance that is instantiating the service is passed into the factory as the first positional argument.

  • enter (bool) – Whether to enter context managers if one is returned by factory. Usually you want that, but there are occasions – like database transaction managers – that you want to enter manually.

  • ping (Callable | None) –

    A callable that marks the service as having a health check.

  • on_registry_close (Callable | Awaitable | None) –

    A callable that is called when the svcs.Registry.close() method is called.

    Can also be an async callable or an collections.abc.Awaitable; then svcs.Registry.aclose() must be called.

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

Syntactic sugar for:

register_factory(
    svc_type,
    lambda: value,
    enter=enter,
    ping=ping,
    on_registry_close=on_registry_close
)
class svcs.Container#

A per-context container for instantiated services and cleanups.

The instance of this should live as long as a request or a task.

Also works as a context manager that runs clean ups on exit:

>>> reg = svcs.Registry()
>>> def factory() -> str:
...     yield "Hello World"
...     print("Cleaned up!")
>>> reg.register_factory(str, factory)

>>> with svcs.Container(reg) as con:
...     _ = con.get(str)
Cleaned up!
Warns:

ResourceWarning – If a container with pending cleanups is garbage-collected.

registry#

The Registry instance that this container uses for service type lookup.

Type:

Registry

__contains__(svc_type)[source]#

Check whether this container has a cached instance of svc_type.

async aclose()[source]#

Run all registered cleanups – synchronous and asynchronous.

Errors are logged at warning level, but otherwise ignored.

Also works with synchronous services, so in an async application, just use this.

Hint

The Container can be used again after this. Closing it is an idempotent way to reset it.

async aget(*svc_types)[source]#

Same as get() but instantiates asynchronously, if necessary.

Also works with synchronous services, so in an async application, just use this.

async aget_abstract(*svc_types)[source]#

Same as get_abstract() but instantiates asynchronously, if necessary.

Also works with synchronous services, so in an async application, just use this.

close()[source]#

Run all registered synchronous cleanups.

Async closes are not awaited and a warning is raised.

Errors are logged at warning level, but otherwise ignored.

Hint

The Container can be used again after this. Closing it is an idempotent way to reset it.

get(*svc_types)[source]#

Get services of svc_types.

Instantiate them if necessary and register their cleanup.

Returns:

If one service is requested, it’s returned directly. If multiple are requested, a tuple of services is returned.

Return type:

svc_types[0] | tuple[*svc_types]

get_abstract(*svc_types)[source]#

Like get() but is annotated to return typing.Any which allows it to be used with abstract types like typing.Protocol or abc classes.

Note

See Typing Caveats why this is necessary.

get_pings()[source]#

Return all services that have defined a ping and bind them to this container.

Returns:

A list of services that have registered a ping callable.

Return type:

list[ServicePing]

class svcs.ServicePing#

A service health check as returned by svcs.Container.get_pings().

name#

A fully-qualified name of the service type.

Type:

str

is_async#

Whether the service needs to be pinged using aping().

Type:

bool

See also

Health Checks

async aping()[source]#

Same as ping() but instantiate and/or ping asynchronously, if necessary.

Also works with synchronous services, so in an async application, just use this.

ping()[source]#

Instantiate the service, schedule its cleanup, and call its ping method.

class svcs.exceptions.ServiceNotFoundError[source]#

Raised when the requested service type is not registered.