audit_trail.context

Request context: who, where and how an audited change happened.

The context reaches the audit writer in one of three ways, checked in this order by :func:resolve_context:

1. bound to a session with :func:bind (session.info["audit_context"]);
2. returned by a host-supplied context_provider;
3. the built-in context variable, set with :func:context.

When none of them yields a context, an anonymous one is used.

Actor dataclass

Actor(
    type: str,
    id: str | None = None,
    label: str | None = None,
)

Who performed an action.

Attributes:

Name	Type	Description
type	str	Kind of actor, defined by the host (for example "user", "system", "api_key").
id	str | None	Opaque identifier of the actor, stored in actor_id.
label	str | None	Human-readable snapshot (for example an e-mail address) that survives deletion of the actor.

AuditContext dataclass

AuditContext(
    actor_type: str = _ANONYMOUS,
    actor_id: str | None = None,
    actor_label: str | None = None,
    remote_addr: str | None = None,
    user_agent: str | None = None,
    method: str | None = None,
    path: str | None = None,
    channel: str | None = None,
    auth_method: str | None = None,
    request_id: UUID | None = None,
    correlation_id: UUID | None = None,
    scope_id: str | None = None,
    extra: dict[str, Any] = dict(),
)

Who, where and how of the current unit of work.

The object is mutable on purpose: it is set once (for example by a middleware) and later updated in place with :func:set_actor, so changes made in a worker thread are visible to the code that started it.

Attributes:

Name	Type	Description
actor_type	str	Kind of actor. Defaults to "anonymous".
actor_id	str | None	Opaque identifier of the actor.
actor_label	str | None	Human-readable snapshot of the actor.
remote_addr	str | None	Client IP address.
user_agent	str | None	Client user agent.
method	str | None	Request method, for example "POST".
path	str | None	Request path.
channel	str | None	Entry point, for example "api", "admin", "cli", "worker".
auth_method	str | None	How the actor authenticated.
request_id	UUID | None	Identifier of the request.
correlation_id	UUID | None	Links several database transactions. When not given, it is set to request_id at construction; changing request_id later does not update it.
scope_id	str | None	Scope of the change, for example a tenant.
extra	dict[str, Any]	Additional host-defined context, stored as meta.

ContextScope

ContextScope(ctx: AuditContext)

Sync and async context manager that activates one AuditContext.

Returned by :func:context; entering it yields the activated context.

Parameters:

Name	Type	Description	Default
ctx	AuditContext	The context to activate.	required

Source code in audit_trail/context.py

def __init__(self, ctx: AuditContext) -> None:
    self._ctx = ctx
    self._tokens: list[Token[AuditContext | None]] = []

ContextSnapshot

Bases: TypedDict

The data.context snapshot stored on each audit entry.

Every key is optional: empty values are left out.

current_context

current_context() -> AuditContext | None

Return the context set with :func:context, if any.

Returns:

Type	Description
AuditContext | None	The active context object, or None outside a :func:context block.

Source code in audit_trail/context.py

def current_context() -> AuditContext | None:
    """Return the context set with :func:`context`, if any.

    Returns:
        The active context object, or ``None`` outside a :func:`context` block.
    """
    return _current.get()

context

context(
    ctx: AuditContext | None = None,
    /,
    *,
    actor_type: str | None = None,
    actor_id: str | None = None,
    actor_label: str | None = None,
    remote_addr: str | None = None,
    user_agent: str | None = None,
    method: str | None = None,
    path: str | None = None,
    channel: str | None = None,
    auth_method: str | None = None,
    request_id: UUID | None = None,
    correlation_id: UUID | None = None,
    scope_id: str | None = None,
    extra: dict[str, Any] | None = None,
) -> ContextScope

Activate an audit context for a block of code.

Usable as with context(...) and async with context(...). The previous context is restored on exit. A nested block replaces the outer context; it does not inherit its fields.

Pass either a ready :class:AuditContext or the fields of a new one; the keyword arguments mirror the :class:AuditContext attributes.

Example::

with context(actor_type="system", actor_label="cleanup", channel="worker"):
    ...

Parameters:

Name	Type	Description	Default
ctx	AuditContext | None	A ready context to activate. Mutually exclusive with the keyword arguments.	None
actor_type	str | None	Kind of actor. None means "anonymous".	None
actor_id	str | None	Opaque identifier of the actor.	None
actor_label	str | None	Human-readable snapshot of the actor.	None
remote_addr	str | None	Client IP address.	None
user_agent	str | None	Client user agent.	None
method	str | None	Request method.	None
path	str | None	Request path.	None
channel	str | None	Entry point, for example "worker".	None
auth_method	str | None	How the actor authenticated.	None
request_id	UUID | None	Identifier of the request.	None
correlation_id	UUID | None	Links several database transactions. Defaults to request_id.	None
scope_id	str | None	Scope of the change, for example a tenant.	None
extra	dict[str, Any] | None	Additional host-defined context, stored as meta.	None

Returns:

Type	Description
ContextScope	A context manager yielding the active :class:AuditContext.

Raises:

Type	Description
TypeError	If both ctx and keyword arguments are given.

Source code in audit_trail/context.py

def context(
    ctx: AuditContext | None = None,
    /,
    *,
    actor_type: str | None = None,
    actor_id: str | None = None,
    actor_label: str | None = None,
    remote_addr: str | None = None,
    user_agent: str | None = None,
    method: str | None = None,
    path: str | None = None,
    channel: str | None = None,
    auth_method: str | None = None,
    request_id: UUID | None = None,
    correlation_id: UUID | None = None,
    scope_id: str | None = None,
    extra: dict[str, Any] | None = None,
) -> ContextScope:
    """Activate an audit context for a block of code.

    Usable as ``with context(...)`` and ``async with context(...)``. The
    previous context is restored on exit. A nested block replaces the outer
    context; it does not inherit its fields.

    Pass either a ready :class:`AuditContext` or the fields of a new one; the
    keyword arguments mirror the :class:`AuditContext` attributes.

    Example::

        with context(actor_type="system", actor_label="cleanup", channel="worker"):
            ...

    Args:
        ctx: A ready context to activate. Mutually exclusive with the
            keyword arguments.
        actor_type: Kind of actor. ``None`` means ``"anonymous"``.
        actor_id: Opaque identifier of the actor.
        actor_label: Human-readable snapshot of the actor.
        remote_addr: Client IP address.
        user_agent: Client user agent.
        method: Request method.
        path: Request path.
        channel: Entry point, for example ``"worker"``.
        auth_method: How the actor authenticated.
        request_id: Identifier of the request.
        correlation_id: Links several database transactions. Defaults to
            ``request_id``.
        scope_id: Scope of the change, for example a tenant.
        extra: Additional host-defined context, stored as ``meta``.

    Returns:
        A context manager yielding the active :class:`AuditContext`.

    Raises:
        TypeError: If both ``ctx`` and keyword arguments are given.
    """
    fields = (
        actor_type,
        actor_id,
        actor_label,
        remote_addr,
        user_agent,
        method,
        path,
        channel,
        auth_method,
        request_id,
        correlation_id,
        scope_id,
        extra,
    )
    if ctx is not None:
        if any(value is not None for value in fields):
            raise TypeError("pass either an AuditContext or its fields, not both")
        return ContextScope(ctx)
    return ContextScope(
        AuditContext(
            actor_type=actor_type if actor_type is not None else _ANONYMOUS,
            actor_id=actor_id,
            actor_label=actor_label,
            remote_addr=remote_addr,
            user_agent=user_agent,
            method=method,
            path=path,
            channel=channel,
            auth_method=auth_method,
            request_id=request_id,
            correlation_id=correlation_id,
            scope_id=scope_id,
            extra=extra if extra is not None else {},
        )
    )

set_actor

set_actor(actor: Actor) -> AuditContext

Set the actor on the active context, in place.

The active object is mutated rather than replaced, so an actor set from a thread pool (for example a synchronous dependency) is visible to the caller.

A context bound with :func:bind takes precedence, so set_actor on the active object does not reach that session; set the actor fields on the bound object instead.

Parameters:

Name	Type	Description	Default
actor	Actor	The actor. All three actor fields are overwritten.	required

Returns:

Type	Description
AuditContext	The updated context.

Raises:

Type	Description
RuntimeError	If no :func:context block is active.

Source code in audit_trail/context.py

def set_actor(actor: Actor) -> AuditContext:
    """Set the actor on the active context, in place.

    The active object is mutated rather than replaced, so an actor set from a
    thread pool (for example a synchronous dependency) is visible to the
    caller.

    A context bound with :func:`bind` takes precedence, so ``set_actor`` on
    the active object does not reach that session; set the actor fields on
    the bound object instead.

    Args:
        actor: The actor. All three actor fields are overwritten.

    Returns:
        The updated context.

    Raises:
        RuntimeError: If no :func:`context` block is active.
    """
    ctx = _current.get()
    if ctx is None:
        raise RuntimeError(
            "no active audit context: enter context() first, or bind() an "
            "AuditContext to the session and set its actor fields directly"
        )
    ctx.actor_type = actor.type
    ctx.actor_id = actor.id
    ctx.actor_label = actor.label
    return ctx

bind

bind(
    session: Session | AsyncSession, context: AuditContext
) -> None

Attach a context to one session.

Takes precedence over the provider and the built-in context variable. Binding does not enable auditing on the session.

Parameters:

Name	Type	Description	Default
session	Session | AsyncSession	The session to attach the context to.	required
context	AuditContext	The context used for audit entries written by this session.	required

Source code in audit_trail/context.py

def bind(session: Session | AsyncSession, context: AuditContext) -> None:
    """Attach a context to one session.

    Takes precedence over the provider and the built-in context variable.
    Binding does not enable auditing on the session.

    Args:
        session: The session to attach the context to.
        context: The context used for audit entries written by this session.
    """
    session.info[_SESSION_INFO_KEY] = context

resolve_context

resolve_context(
    session: Session | AsyncSession,
    context_provider: Callable[[], AuditContext | None]
    | None = None,
) -> AuditContext

Return the context that applies to session.

Checks, in order: the context bound with :func:bind, then context_provider(), then the built-in context variable. The object is returned as is, not copied, so later :func:set_actor calls stay visible.

Parameters:

Name	Type	Description	Default
session	Session | AsyncSession	The session being audited.	required
context_provider	Callable[[], AuditContext | None] | None	Optional host hook returning the current context, or None to fall through.	None

Returns:

Type	Description
AuditContext	The first context found, or a new anonymous one.

Raises:

Type	Description
TypeError	If session.info["audit_context"] is not an :class:AuditContext.

Source code in audit_trail/context.py

def resolve_context(
    session: Session | AsyncSession,
    context_provider: Callable[[], AuditContext | None] | None = None,
) -> AuditContext:
    """Return the context that applies to ``session``.

    Checks, in order: the context bound with :func:`bind`, then
    ``context_provider()``, then the built-in context variable. The object is
    returned as is, not copied, so later :func:`set_actor` calls stay visible.

    Args:
        session: The session being audited.
        context_provider: Optional host hook returning the current context,
            or ``None`` to fall through.

    Returns:
        The first context found, or a new anonymous one.

    Raises:
        TypeError: If ``session.info["audit_context"]`` is not an
            :class:`AuditContext`.
    """
    bound = session.info.get(_SESSION_INFO_KEY)
    if bound is not None:
        if not isinstance(bound, AuditContext):
            raise TypeError(
                f'session.info["{_SESSION_INFO_KEY}"] must be an AuditContext, '
                f"got {type(bound).__name__}"
            )
        return bound
    if context_provider is not None:
        provided = context_provider()
        if provided is not None:
            return provided
    active = _current.get()
    if active is not None:
        return active
    return AuditContext()

context_snapshot

context_snapshot(ctx: AuditContext) -> ContextSnapshot

Build the data.context snapshot stored on each audit entry.

Empty values (None, "", an empty extra) are left out. actor_id is never included: it is stored in its own column.

Parameters:

Name	Type	Description	Default
ctx	AuditContext	The context to snapshot.	required

Returns:

Type	Description
ContextSnapshot	The snapshot; request_id is a string and extra is copied as
ContextSnapshot	meta. meta values are encoded to JSON by the writer.

Source code in audit_trail/context.py

def context_snapshot(ctx: AuditContext) -> ContextSnapshot:
    """Build the ``data.context`` snapshot stored on each audit entry.

    Empty values (``None``, ``""``, an empty ``extra``) are left out.
    ``actor_id`` is never included: it is stored in its own column.

    Args:
        ctx: The context to snapshot.

    Returns:
        The snapshot; ``request_id`` is a string and ``extra`` is copied as
        ``meta``. ``meta`` values are encoded to JSON by the writer.
    """
    snapshot: ContextSnapshot = {}
    if ctx.actor_type:
        snapshot["actor_type"] = ctx.actor_type
    if ctx.actor_label:
        snapshot["actor_label"] = ctx.actor_label
    if ctx.remote_addr:
        snapshot["remote_addr"] = ctx.remote_addr
    if ctx.user_agent:
        snapshot["user_agent"] = ctx.user_agent
    if ctx.method:
        snapshot["method"] = ctx.method
    if ctx.path:
        snapshot["path"] = ctx.path
    if ctx.channel:
        snapshot["channel"] = ctx.channel
    if ctx.auth_method:
        snapshot["auth_method"] = ctx.auth_method
    if ctx.request_id is not None:
        snapshot["request_id"] = str(ctx.request_id)
    if ctx.extra:
        snapshot["meta"] = dict(ctx.extra)
    return snapshot