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.

In 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

>>> registry = svcs.Registry()

>>> registry.register_factory(uuid.UUID, uuid.uuid4)
>>> registry.register_value(str, "Hello World")
>>> uuid.UUID in registry
True
>>> str in registry
True
>>> int in registry
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.

Multiple Factories for the Same Type#

Sometimes, it makes sense to have multiple instances of the same type. For example, you might have multiple HTTP client pools or more than one database connection.

You can achieve this by using either typing.Annotated (Python 3.9+, or in typing-extensions) or by using type (Python 3.12+, use typing.NewType on older versions). You can also mix and match the two. For instance, if you need a primary and a secondary database connection:

from typing import Annotated, NewType

from sqlalchemy import Connection, create_engine

# Create the two connection engines
primary_engine = create_engine(primary_url)
secondary_engine = create_engine(secondary_url)

# Create unique types for both with two different approaches
PrimaryConnection = Annotated[Connection, "primary"]
SecondaryConnection = NewType("SecondaryConnection", Connection)
# Or on Python 3.12:
# type SecondaryConnection = Connection

# Register the factories to the aliases
registry.register_factory(PrimaryConnection, primary_engine.connect)
registry.register_factory(SecondaryConnection, secondary_engine.connect)

The type and content of the annotated metadata (“primary”) are not important to svcs, as long as the whole type is hashable.

Cleanup#

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

engine = create_engine(url)

registry.register_factory(
    Connection, engine.connect, 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 when you call its get() or aget() method:

>>> container = svcs.Container(registry)

>>> 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.

Our Integrations offer a svcs_from() function to extract the container from the current environment, and a get() (and/or aget()) function that transparently gets the service from the current container for you. Depending on your web framework, you may have to pass the current request object as the first argument to svcs_from() / get() / aget().

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(registry)

# Let's make the UUID predictable for our test!
>>> registry.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

>>> registry.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. Of course, you can also use Container-Local Registries.

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 disable this behavior by passing enter=False to register_factory() and it’s off by default for 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 clean_factory() -> str:
...     yield "Hello World"
...     print("Cleaned up!")
>>> reg.register_factory(str, clean_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.

Container-Local Registries#

New in version 23.21.0.

Sometimes, you want to register a factory or value that’s only valid within a container. For example, you might want to register a factory that depends on data from a request object. Per-request factories, if you will.

This is where container-local registries come in. They are created implicitly by calling svcs.Container.register_local_factory() and svcs.Container.register_local_value(). When looking up factories in a container, the local registry takes precedence over the global one, and it is closed along with the container:

>>> container = svcs.Container(registry)
>>> registry.register_value(str, "Hello World!")
>>> container.register_local_value(str, "Goodbye Cruel World!")
>>> container.get(str)
'Goodbye Cruel World!'
>>> container.close()  # closes both container & its local registry
>>> registry.close()   # closes the global registry

Warning

Nothing is going to stop you from letting your global factories depend on local ones – similarly to template subclassing.

For example, you could define your database connection like this:

from sqlalchemy import text

def connect_and_set_user(svcs_container):
    user_id = svcs_container.get(User).user_id
    with engine.connect() as conn:
        conn.execute(text("SET user = :id", {"id": user_id}))

        yield conn

registry.register_factory(Connection, connect_and_set_user)

And then, somewhere in a middleware, define a local factory for the Request type using something like:

def middleware(request):
    container.register_local_value(User, User(request.user_id, ...))

However, then you have to be very careful around the caching of created services. If your application requests a Connection instance before you register the local Request factory, the Connection factory will either crash or be created with the wrong user (for example, if you defined a stub/fallback user in the global registry).

It is safer and easier to reason about your code if you keep the dependency arrows point from the local registry to the global one:

# The global connection factory that creates and cleans up vanilla
# connections.
registry.register_factory(Connection, engine.connect)

# Create a type alias with an idiomatic name.
type ConnectionWithUserID = Connection

def middleware(request):
    def set_user_id(svcs_container):
        conn = svcs_container.get(Connection)
        conn.execute(text("SET user = :id", {"id": user_id}))

        return conn

    # Use a factory to keep the service lazy. If the view never asks for a
    # connection, we never connect -- or set a user.
    container.register_local_factory(ConnectionWithUserID, set_user_id)

Now the type name expresses the purpose of the object and it doesn’t matter if there’s already a non-user-aware Connection in the global registry.

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:

AIOHTTP

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)

FastAPI

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
    )

Flask

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

Pyramid

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"),
    )

Starlette

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:

The svcs.Registry objects live on application-scoped objects like flask.Flask.config.
The svcs.Container objects live on request-scoped objects like flask.g.

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

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]#

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.

Note

Generally speaking, given the churn and edgecases in the typing ecosystem, we recommend using the name route to detect the container argument because it’s most reliable.
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.

See also

Container.get_pings() and ServicePing.
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=False, 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
)

Please note that, unlike with register_factory(), entering context managers is disabled by default.

Changed in version 23.21.0: enter is now False by default.

class svcs.Container[source]#

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:: svcs._core.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_type: type[T1], /) → T1[source]#

async aget(svc_type1: type[T1], svc_type2: type[T2], /) → tuple[T1, T2]

async aget(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], /) → tuple[T1, T2, T3]

async aget(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], /) → tuple[T1, T2, T3, T4]

async aget(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], /) → tuple[T1, T2, T3, T4, T5]

async aget(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], svc_type6: type[T6], /) → tuple[T1, T2, T3, T4, T5, T6]

async aget(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], svc_type6: type[T6], svc_type7: type[T7], /) → tuple[T1, T2, T3, T4, T5, T6, T7]

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_type: type[T1], /) → T1[source]#

get(svc_type1: type[T1], svc_type2: type[T2], /) → tuple[T1, T2]

get(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], /) → tuple[T1, T2, T3]

get(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], /) → tuple[T1, T2, T3, T4]

get(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], /) → tuple[T1, T2, T3, T4, T5]

get(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], svc_type6: type[T6], /) → tuple[T1, T2, T3, T4, T5, T6]

get(svc_type1: type[T1], svc_type2: type[T2], svc_type3: type[T3], svc_type4: type[T4], svc_type5: type[T5], svc_type6: type[T6], svc_type7: type[T7], /) → tuple[T1, T2, T3, T4, T5, T6, T7]

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]

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

Same as svcs.Registry.register_factory(), but registers the factory only for this container.

A temporary svcs.Registry is transparently created – and closed together with the container it belongs to.