Source code for svcs.fastapi

# SPDX-FileCopyrightText: 2023 Hynek Schlawack <hs@ox.cx>
#
# SPDX-License-Identifier: MIT

from __future__ import annotations

import contextlib
import inspect

from collections.abc import AsyncGenerator, Callable
from typing import Annotated, TypeAlias, cast

import attrs

from fastapi import Depends, FastAPI, Request

import svcs

from svcs._core import _KEY_REGISTRY


AsyncGenLifespan: TypeAlias = Callable[
    [FastAPI, svcs.Registry],
    AsyncGenerator[dict[str, object] | None, None],
]
AsyncCMLifespan: TypeAlias = Callable[
    [FastAPI, svcs.Registry],
    contextlib.AbstractAsyncContextManager[dict[str, object] | None],
]
SomeLifespan: TypeAlias = AsyncGenLifespan | AsyncCMLifespan




@attrs.define
class lifespan:  # noqa: N801
    """
    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.

    Args:
        lifespan: The lifespan function to make *svcs*-aware.
    """

    _lifespan: SomeLifespan
    _state: dict[str, object] = attrs.field(factory=dict)
    registry: svcs.Registry = attrs.field(factory=svcs.Registry)

    @contextlib.asynccontextmanager
    async def __call__(
        self, app: FastAPI
    ) -> AsyncGenerator[dict[str, object], None]:
        cm: AsyncCMLifespan
        if inspect.isasyncgenfunction(self._lifespan):
            cm = contextlib.asynccontextmanager(
                cast(AsyncGenLifespan, self._lifespan)
            )
        else:
            cm = cast(AsyncCMLifespan, self._lifespan)

        async with self.registry, cm(app, self.registry) as state:
            self._state = state or {}
            self._state[_KEY_REGISTRY] = self.registry
            yield self._state





async def container(request: Request) -> AsyncGenerator[svcs.Container, None]:
    """
    A FastAPI `dependency
    <https://fastapi.tiangolo.com/tutorial/dependencies/>`_ that provides you
    with a request-scoped container.

    Yields:
        A :class:`svcs.Container` that is cleaned up after the request.
    """
    async with svcs.Container(getattr(request.state, _KEY_REGISTRY)) as cont:
        yield cont



DepContainer = Annotated[svcs.Container, Depends(container)]
"""
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):
        ...
"""