audit_trail.integrations.fastapi

FastAPI integration: ASGI middleware, session dependency and providers.

Needs the fastapi extra (pip install 'sqlalchemy-audit-trail[fastapi]').

:class:AuditMiddleware activates an AuditContext for each request, with the network part filled in: client address, user agent, method, path, channel and request id. The same object stays active for the whole request, including synchronous dependencies and endpoints, which FastAPI runs in its thread pool in a copy of the request's context variables. The copy still refers to the same object, so :func:set_actor called there updates the context the session listener reads when it flushes.

Example::

from audit_trail import Actor
from audit_trail.integrations.fastapi import (
    AuditMiddleware,
    session_dependency,
    set_actor,
)

app = FastAPI()
app.add_middleware(AuditMiddleware)
get_session = session_dependency(SessionLocal)  # the factory given to install()


def current_user(token: str = Depends(oauth2_scheme)) -> User:
    user = authenticate(token)
    set_actor(Actor("user", str(user.id), user.email), auth_method="bearer")
    return user

AuditMiddleware

AuditMiddleware(
    app: ASGIApp,
    *,
    channel: str | None = "api",
    request_id_header: str | None = "x-request-id",
    trusted_proxies: Iterable[str] = (),
)

ASGI middleware that activates an audit context for each request.

For http and websocket requests it builds an AuditContext with remote_addr, user_agent, method (None for a websocket), path (without the query string), channel and request_id, and runs the application inside context(...). Other scopes, such as lifespan, pass through untouched.

Headers set by clients are trusted only when the direct peer is a trusted proxy; by default none is:

• remote_addr is the direct peer's address. When the peer is in trusted_proxies, it is taken from X-Forwarded-For instead: the list is read from right to left, trusted proxies are skipped and the first other address is the client (the leftmost one if all are trusted). Forwarded (RFC 7239) is not read.
• request_id comes from the request_id_header only when the peer is in trusted_proxies and the value is a UUID; otherwise a new UUID is generated, so a client cannot reuse the id of another request.

An address that is not a valid IP (for example Starlette's "testclient") is stored as None: remote_addr is an inet column.

Parameters:

Name	Type	Description	Default
app	ASGIApp	The ASGI application to wrap.	required
channel	str | None	Stored as the context's channel.	'api'
request_id_header	str | None	Header carrying the request id set by a trusted proxy. None always generates a new id.	'x-request-id'
trusted_proxies	Iterable[str]	IP addresses or networks ("10.0.0.0/8") of the proxies whose X-Forwarded-For and request id headers are believed. Empty by default.	()

Raises:

Type	Description
ValueError	An entry of trusted_proxies is not an IP address or network.

Source code in audit_trail/integrations/fastapi.py

def __init__(
    self,
    app: ASGIApp,
    *,
    channel: str | None = "api",
    request_id_header: str | None = "x-request-id",
    trusted_proxies: Iterable[str] = (),
) -> None:
    self.app = app
    self.channel = channel
    self.request_id_header = (
        None
        if request_id_header is None
        else request_id_header.lower().encode("latin-1")
    )
    self.trusted_proxies: tuple[_Network, ...] = tuple(
        ipaddress.ip_network(entry, strict=False) for entry in trusted_proxies
    )

request_context

request_context() -> AuditContext | None

Return the context AuditMiddleware activated for this request.

A nested audit.context(...) block does not change it.

Returns:

Type	Description
AuditContext | None	The request's context, or None outside a request.

Source code in audit_trail/integrations/fastapi.py

def request_context() -> AuditContext | None:
    """Return the context ``AuditMiddleware`` activated for this request.

    A nested ``audit.context(...)`` block does not change it.

    Returns:
        The request's context, or ``None`` outside a request.
    """
    state = _request.get()
    return None if state is None else state.context

context_provider

context_provider() -> AuditContext | None

context_provider for AuditTrail that follows the active context.

Returns the context of an audit.context(...) block when one is active (a nested block inside a request wins), otherwise the request's context. Passing it is optional when AuditMiddleware is installed: the middleware activates the request's context in the built-in context variable, which the listener reads anyway.

Returns:

Type	Description
AuditContext | None	The active context, or None outside a request and any block.

Source code in audit_trail/integrations/fastapi.py

def context_provider() -> AuditContext | None:
    """``context_provider`` for ``AuditTrail`` that follows the active context.

    Returns the context of an ``audit.context(...)`` block when one is
    active (a nested block inside a request wins), otherwise the request's
    context. Passing it is optional when ``AuditMiddleware`` is installed:
    the middleware activates the request's context in the built-in context
    variable, which the listener reads anyway.

    Returns:
        The active context, or ``None`` outside a request and any block.
    """
    active = current_context()
    return active if active is not None else request_context()

session_provider

session_provider() -> Session | AsyncSession | None

session_provider for AuditTrail: this request's session.

Returns:

Type	Description
Session | AsyncSession | None	The session opened by a :func:session_dependency dependency for
Session | AsyncSession | None	the current request, or None when there is none or no
Session | AsyncSession | None	AuditMiddleware is installed.

Source code in audit_trail/integrations/fastapi.py

def session_provider() -> Session | AsyncSession | None:
    """``session_provider`` for ``AuditTrail``: this request's session.

    Returns:
        The session opened by a :func:`session_dependency` dependency for
        the current request, or ``None`` when there is none or no
        ``AuditMiddleware`` is installed.
    """
    state = _request.get()
    return None if state is None else state.session

set_actor

set_actor(
    actor: Actor, *, auth_method: str | None = None
) -> AuditContext

Set the actor of the current request, after authentication.

Updates the request's context (the one AuditMiddleware activated) in place, not a nested audit.context(...) block, so it can be called from a synchronous dependency running in the thread pool. Entries flushed afterwards in the request carry the actor.

Parameters:

Name	Type	Description	Default
actor	Actor	The actor. All three actor fields are overwritten.	required
auth_method	str | None	How the actor authenticated, stored when given.	None

Returns:

Type	Description
AuditContext	The request's context.

Raises:

Type	Description
RuntimeError	No AuditMiddleware request is active.

Source code in audit_trail/integrations/fastapi.py

def set_actor(actor: Actor, *, auth_method: str | None = None) -> AuditContext:
    """Set the actor of the current request, after authentication.

    Updates the request's context (the one ``AuditMiddleware`` activated) in
    place, not a nested ``audit.context(...)`` block, so it can be called
    from a synchronous dependency running in the thread pool. Entries flushed
    afterwards in the request carry the actor.

    Args:
        actor: The actor. All three actor fields are overwritten.
        auth_method: How the actor authenticated, stored when given.

    Returns:
        The request's context.

    Raises:
        RuntimeError: No ``AuditMiddleware`` request is active.
    """
    ctx = request_context()
    if ctx is None:
        raise RuntimeError(
            "no active request context: add AuditMiddleware to the application"
        )
    ctx.actor_type = actor.type
    ctx.actor_id = actor.id
    ctx.actor_label = actor.label
    if auth_method is not None:
        ctx.auth_method = auth_method
    return ctx

session_dependency

session_dependency(
    factory: sessionmaker[_S],
) -> Callable[[], Generator[_S, None, None]]

session_dependency(
    factory: async_sessionmaker[_A],
) -> Callable[[], AsyncGenerator[_A, None]]

session_dependency(
    factory: sessionmaker[_S] | async_sessionmaker[_A],
) -> (
    Callable[[], Generator[_S, None, None]]
    | Callable[[], AsyncGenerator[_A, None]]
)

Build a FastAPI dependency yielding a session of factory.

Pass the factory given to AuditTrail.install. The session is closed after the request and never committed by the dependency. For a sessionmaker the dependency is synchronous, so FastAPI runs it in its thread pool; for an async_sessionmaker it is asynchronous.

The session is not bound to a context: the listener reads the request's context (or a nested audit.context(...)) when it flushes, so an actor set with :func:set_actor after the session was opened still reaches its entries.

Parameters:

Name	Type	Description	Default
factory	sessionmaker[_S] | async_sessionmaker[_A]	A sessionmaker or async_sessionmaker.	required

Returns:

Type	Description
Callable[[], Generator[_S, None, None]] | Callable[[], AsyncGenerator[_A, None]]	The dependency, for Depends(...). It raises TypeError when
Callable[[], Generator[_S, None, None]] | Callable[[], AsyncGenerator[_A, None]]	the session's class has no AuditTrail installed.

Raises:

Type	Description
TypeError	factory is neither.

Source code in audit_trail/integrations/fastapi.py

def session_dependency(
    factory: sessionmaker[_S] | async_sessionmaker[_A],
) -> Callable[[], Generator[_S, None, None]] | Callable[[], AsyncGenerator[_A, None]]:
    """Build a FastAPI dependency yielding a session of ``factory``.

    Pass the factory given to ``AuditTrail.install``. The session is closed
    after the request and never committed by the dependency. For a
    ``sessionmaker`` the dependency is synchronous, so FastAPI runs it in
    its thread pool; for an ``async_sessionmaker`` it is asynchronous.

    The session is not bound to a context: the listener reads the request's
    context (or a nested ``audit.context(...)``) when it flushes, so an actor
    set with :func:`set_actor` after the session was opened still reaches
    its entries.

    Args:
        factory: A ``sessionmaker`` or ``async_sessionmaker``.

    Returns:
        The dependency, for ``Depends(...)``. It raises ``TypeError`` when
        the session's class has no ``AuditTrail`` installed.

    Raises:
        TypeError: ``factory`` is neither.
    """
    if isinstance(factory, sessionmaker):
        sync_factory: sessionmaker[_S] = factory

        def get_session() -> Generator[_S, None, None]:
            with sync_factory() as session:
                _check_installed(session)
                with _tracked(session):
                    yield session

        return get_session

    from sqlalchemy.ext.asyncio import async_sessionmaker

    if not isinstance(factory, async_sessionmaker):
        raise TypeError(
            "session_dependency() needs a sessionmaker or an async_sessionmaker, "
            f"got {factory!r}"
        )
    async_factory: async_sessionmaker[_A] = factory

    async def get_async_session() -> AsyncGenerator[_A, None]:
        async with async_factory() as session:
            _check_installed(session.sync_session)
            with _tracked(session):
                yield session

    return get_async_session