FastAPI¶

svcs’s centralization and on-demand capabilities are a great complement to FastAPI’s dependency injection system – especially when the annotated decorator approach becomes unwieldy because of too many dependencies.

svcs’s FastAPI integration stores the svcs.Registry on the lifespan state and the svcs.Container is a dependency that can be injected into your views. That makes very little API necessary from svcs itself.

Initialization¶

FastAPI inherited the request.state attribute from starlette and svcs uses it to store the svcs.Registry on it.

To get it there you have to instantiate your FastAPI application with a lifespan. Whatever this lifespan yields, becomes the initial request state via shallow copy.

To keep track of its registry for later overwriting, svcs comes with the svcs.fastapi.lifespan decorator that remembers the registry on the lifespan object (see below in testing to see it in action):

from fastapi import FastAPI

import svcs

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

    yield {"your": "other", "initial": "state"}

    # Registry is closed automatically when the app is done.

app = FastAPI(lifespan=lifespan)

Service Acquisition¶

svcs comes with the svcs.fastapi.container() dependency that will inject a request-scoped svcs.Container into your views if the application is correctly initialized:

from typing import Annotated

from fastapi import Depends


@app.get("/")
async def index(services: Annotated[svcs.Container, Depends(svcs.fastapi.container)]):
    db = services.get(Database)

For your convenience, svcs comes with the alias svcs.fastapi.DepContainer that allows you to use the shorter and even nicer:

@app.get("/")
async def index(services: svcs.fastapi.DepContainer):
    db = services.get(Database)

Health Checks¶

With the help of the svcs.fastapi.container() dependency you can easily add a health check endpoint to your application without any special API:

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
    )

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 FastAPI application as an example:

from __future__ import annotations

import os

from typing import AsyncGenerator

from fastapi import FastAPI

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


@svcs.fastapi.lifespan
async def lifespan(
    app: FastAPI, 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 = FastAPI(lifespan=lifespan)


@app.get("/users/{user_id}")
async def get_user(user_id: int, services: svcs.fastapi.DepContainer) -> dict:
    db = await services.aget(Database)

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

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 fastapi.testclient import TestClient

from simple_fastapi_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 as shown above, and use the svcs.fastapi.container() dependency to get your services, everything is cleaned up behind you automatically.

API Reference¶

Application Life Cycle¶

class svcs.fastapi.lifespan(lifespan)[source]¶

Make a FastAPI 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[[FastAPI, svcs.Registry], contextlib.AbstractAsyncContextManager[dict[str, object]]] | Callable[[FastAPI, svcs.Registry], contextlib.AbstractAsyncContextManager[None]] | Callable[[FastAPI, svcs.Registry], AsyncGenerator[dict[str, object], None]] | Callable[[FastAPI, svcs.Registry], AsyncGenerator[None, None]]) – The lifespan function to make svcs-aware.

Service Acquisition¶

async svcs.fastapi.container(request)[source]¶

A FastAPI dependency that provides you with a request-scoped container.

Yields:: A svcs.Container that is cleaned up after the request.

svcs.fastapi.DepContainer¶

An alias for:

typing.Annotated[svcs.Container, fastapi.Depends(svcs.fastapi.container)]

This allows you write your view like:

@app.get("/")
async def view(services: svcs.fastapi.DepContainer):
    ...