Starlette

svcs’s Starlette integration stores the svcs.Registry on the lifespan state and the svcs.Container is added to the request state using a pure ASGI middleware.

It’s a great way to get type-safety and rich life cycle management on top of Starlette’s low-level dependency capabilities.

Initialization

To use svcs with Starlette, you have to pass a lifespan – that has been wrapped by svcs.starlette.lifespan – and a SVCSMiddleware to your application:

from starlette.applications import Starlette
from starlette.middleware import Middleware

import svcs


@svcs.starlette.lifespan
async def lifespan(app: Starlette, registry: svcs.Registry):
    registry.register_factory(Database, Database.connect)

    yield {"your": "other stuff"}

    # Registry is closed automatically when the app is done.


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[...],
)

Service Acquisition

You can either use svcs.starlette.svcs_from():

from svcs.starlette import svcs_from

async def view(request):
    db = await svcs_from(request).aget(Database)

Or you can use svcs.starlette.aget() to extract your services directly:

import svcs

async def view(request):
    db = await svcs.starlette.aget(request, Database)

Health Checks

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

A health endpoint could look like this:

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
    )

Testing

The centralized service registry makes it straight-forward to selectively replace dependencies within your application in tests even if you have many dependencies to handle.

Let’s take this simple application as an example:

from __future__ import annotations

import os

from typing import AsyncGenerator

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

import svcs


config = {"db_url": os.environ.get("DB_URL", "sqlite:///:memory:")}


class Database:
    @classmethod
    async def connect(cls, db_url: str) -> Database:
        # ...
        return Database()

    async def get_user(self, user_id: int) -> dict[str, str]:
        return {}  # not interesting here


async def get_user(request: Request) -> JSONResponse:
    db = await svcs.starlette.aget(request, Database)

    try:
        return JSONResponse(
            {"data": await db.get_user(request.path_params["user_id"])}
        )
    except Exception as e:
        return JSONResponse({"oh no": e.args[0]})


@svcs.starlette.lifespan
async def lifespan(
    app: Starlette, registry: svcs.Registry
) -> AsyncGenerator[dict[str, object], None]:
    async def connect_to_db() -> Database:
        return await Database.connect(config["db_url"])

    registry.register_factory(Database, connect_to_db)

    yield {"your": "other stuff"}


app = Starlette(
    lifespan=lifespan,
    middleware=[Middleware(svcs.starlette.SVCSMiddleware)],
    routes=[Route("/users/{user_id}", get_user)],
)

Now if you want to make a request against the get_user view, but want the database to raise an error to see if it’s properly handled, you can do this:

from unittest.mock import Mock

import pytest

from starlette.testclient import TestClient

from simple_starlette_app import Database, app, lifespan


@pytest.fixture(name="client")
def _client():
    with TestClient(app) as client:
        yield client


def test_db_goes_boom(client):
    """
    Database errors are handled gracefully.
    """

    # IMPORTANT: Overwriting must happen AFTER the app is ready!
    db = Mock(spec_set=Database)
    db.get_user.side_effect = Exception("boom")
    lifespan.registry.register_value(Database, db)

    resp = client.get("/users/42")

    assert {"oh no": "boom"} == resp.json()

As you can see, we can inspect the decorated lifespan function to get the registry that got injected and you can overwrite it later.

Important

You must overwrite after the application has been initialized. Otherwise the lifespan function overwrites your settings.

Cleanup

If you initialize the application with a lifespan and middleware as shown above, and use svcs_from() or aget() to get your services, everything is cleaned up behind you automatically.

API Reference

Application Life Cycle

class svcs.starlette.lifespan(lifespan)

Make a Starlette lifespan svcs-aware.

Makes sure that the registry is available to the decorated lifespan function as a second parameter and that the registry is closed when the application exists.

Async generators are automatically wrapped into an async context manager.

Parameters:

lifespan (Callable[[Starlette, svcs.Registry], contextlib.AbstractAsyncContextManager[dict[str, object]]] | Callable[[Starlette, svcs.Registry], contextlib.AbstractAsyncContextManager[None]] | Callable[[Starlette, svcs.Registry], AsyncGenerator[dict[str, object], None]] | Callable[[Starlette, svcs.Registry], AsyncGenerator[None, None]]) – The lifespan function to make svcs-aware.

class svcs.starlette.SVCSMiddleware(app)

Attach a svcs.Container to the request state, based on a registry that has been put on the request state by lifespan. Closes the container at the end of a request or websocket connection.

See also

Initialization

Service Acquisition

async svcs.starlette.aget(request: starlette.requests.Request, svc_type1: type, ...)

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

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

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

svcs.starlette.svcs_from(request)

Get the current container from request.

svcs.starlette.get_pings(request)

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

See also

Health Checks