FastAPI lifespan-scoped (singleton) dependencies

example_project.py

########
# app.py
########

# Example of how I use it in my project. This file cannot be run as-is.
# The only difference with the example in the fastapi_singleton module docstring is
# the use of a subclassed FastAPI application to define type annotations
import fastapi
import fastapi_singleton

from . import config, schemas
from .db import Database

router = fastapi.APIRouter()
deps = fastapi_singleton.Dependencies()

class Application(fastapi.FastAPI):
    """Tiny wrapper to add better auto-completion and type-checking on application state."""

    # Set in create_app
    @property
    def conf(self) -> config.ServerConfig:
        return self.state.config

    @conf.setter
    def conf(self, value: config.ServerConfig) -> None:
        self.state.config = value

    # Set in _db
    @property
    def db(self) -> Database:
        return self.state.db

    @db.setter
    def db(self, value: Database) -> None:
        self.state.db = value

def create_app(conf: config.ServerConfig) -> Application:
    app = Application(...)
    app.conf = conf
    deps.connect(app)
    app.include_router(router)
    [...]
    return app

# Dependencies
@deps.singleton
async def _db(app: Application) -> AsyncIterator[Database]:
    """Provide database."""
    db = Database(app.conf)
    db.prepare()
    app.db = db
    yield db
    # no tear-down

[...]

# REST
@router.get("/models", response_model=List[schemas.Model])
def get_models(db: Database = _db) -> List[schemas.Model]:
    """Get available models."""
    return db.get_models()

#########
# asgi.py
#########
import os
from .app import create_app
from .config import read_config

config_file = os.environ["APP_CONFIG_FILE"]
conf = read_config(config_file)
app = create_app(conf)

fastapi_singleton.py

# This file is complete, and should run as-is.
"""
Add support for lifespan-scoped (named singleton in most DI frameworks) dependencies to ``fastapi``.

``fastapi.Depends`` is called on every request. This wraps ``fastapi.FastAPI.on_event("startup"|"shutdown")``
and ``fastapi.Depends``, so as to provide a way to register lifespan-scoped dependencies.

.. code-block:: python

    from typing import AsyncIterator

    import fastapi
    import fastapi_singleton

    # For demo purposes
    class Database:
        def __init__(self, db_url): ...
        async def start(self): ...
        async def shutdown(self): ...

    router = fastapi.APIRouter()
    deps = fastapi_singleton.Dependencies()

    @deps.singleton
    async def get_database(app: fastapi.FastAPI) -> AsyncIterator[Database]:
        config = app.state.config
        db = Database(config.db_url)
        await db.start()
        yield db
        await db.shutdown()

    @router.get("/users")
    def get_users(db: Database = get_database):
        ...

    def create_app(config):
        app = fastapi.FastAPI()
        app.state.config = config
        # ...
        app.include_router(router)
        deps.connect(app)

.. note::
    So as to keep things simple, the implementation has one caveat: the dependency is
    systematically setup/shutdown with the application, regardless of whether its
    dependant is called.

.. seealso::
    - https://github.com/tiangolo/fastapi/issues/504
    - https://github.com/tiangolo/fastapi/issues/617
"""
import contextlib
from typing import Any, AsyncIterator, Callable, Dict

import fastapi

DependencyType = Callable[[Any], AsyncIterator[Any]]

_sentinel_start = object()
_sentinel_shutdown = object()


class Dependencies:
    """Dependency container."""

    def __init__(self) -> None:
        self._registry: Dict[DependencyType, Any] = {}  # dependency function -> value

    def singleton(self, func: DependencyType) -> Any:
        """
        Register a singleton dependency provider.

        :return: A (unresolved) dependency, which can be used wherever the return
            value of ``fastapi.Depends()`` would be used.
        """

        def get_value() -> Any:
            value = self._registry[func]
            if value is _sentinel_start:
                raise RuntimeError("Application not started yet.")
            if value is _sentinel_shutdown:
                raise RuntimeError("Application already shut down.")
            return value

        self._registry[func] = _sentinel_start
        return fastapi.Depends(get_value)

    def connect(self, app: fastapi.FastAPI) -> None:
        """Connect registered dependencies with the application."""
        for dep in self._registry:
            # Defining functions in a for-loop is error-prone (watch out for early binding)
            # and kinda ugly
            self._connect_one(app, dep)

    def _connect_one(self, app: fastapi.FastAPI, dep: DependencyType) -> None:
        # Don't instantiate the context manager yet, otherwise it can only be used once
        cm_factory = contextlib.asynccontextmanager(dep)
        stack = contextlib.AsyncExitStack()

        async def startup() -> None:
            resolved = await stack.enter_async_context(cm_factory(app))
            self._registry[dep] = resolved

        async def shutdown() -> None:
            await stack.pop_all().aclose()
            self._registry[dep] = _sentinel_shutdown

        app.add_event_handler("startup", startup)
        app.add_event_handler("shutdown", shutdown)

test_example_project.py

import contextlib
import pytest
from starlette.testclient import TestClient

from app import create_app
from config import ServerConfig


@pytest.fixture
def mk_client():
    stack = contextlib.ExitStack()
    
    def inner(conf):
        app = create_app(conf)
        client = TestClient(app)
        stack.enter_context(client)
        return client
    
    with stack:
        yield inner
        
        
def test_example(mk_client):
    client = mk_client(ServerConfig(...))
    
    client.post(...)
    [...]
```

test_fastapi_singleton.py

# Complete tests for fastapi_singleton.py
from typing import AsyncIterator, List

import fastapi
from starlette.testclient import TestClient

import fastapi_singleton


class Application(fastapi.FastAPI):
    """Test annotations allow for subclassing without having ``mypy`` bark."""


router = fastapi.APIRouter()
deps = fastapi_singleton.Dependencies()


@deps.singleton
async def get_events(app: Application) -> AsyncIterator[List[str]]:
    events = app.state.events
    events.append("start")
    yield events
    events.append("shutdown")


@router.get("/", response_model=List[str])
def index(events: List[str] = get_events) -> List[str]:
    events.append("/")
    return events


def create_app(events: List[str]) -> Application:
    app = Application()
    app.state.events = events
    app.include_router(router)
    deps.connect(app)
    return app


def test_fastapi_singleton() -> None:
    events: List[str] = []
    app = create_app(events)

    with TestClient(app) as client:
        assert events == ["start"]
        response = client.get("/")
        assert response.json() == ["start", "/"]
        assert events == ["start", "/"]
    assert events == ["start", "/", "shutdown"]

    # ♪ One More Time ♪
    # Make sure the context manager isn't exhausted the first time
    with TestClient(app) as client:
        assert events == ["start", "/", "shutdown", "start"]
    assert events == ["start", "/", "shutdown", "start", "shutdown"]

Thanks for this 😄

Thanks! This should really be a PyPi library!