audit_trail.query

Read path: grouped listing, object history and entry details.

AuditQuery.list_groups lists audit entries grouped by database transaction, newest first, with keyset pagination. It runs in two stages:

1. One query per requested severity (LIMIT limit + 1 each, merged in Python) collects the ids of the page's transactions. A single-severity query reads the monthly partitions in order and stops at the limit; a query over several severities would read all of their rows.
2. The complete groups are fetched with transaction_id IN (...) and explicit created_at bounds, so only the partitions of the page's time range are read, and their audit_transaction rows likewise.

object_history and related page through transactions the same way, with their own stage 1 streams: the object and target indexes for a record's history, the correlation index for related transactions. get reads one entry by (id, created_at), which prunes to one monthly partition per severity (one partition when the severity is given), and access_summary counts an object's accesses per actor. LabelResolver labels the objects that foreign-key and relationship changes refer to.

Every query runs after SET LOCAL plan_cache_mode = 'force_custom_plan': a generic plan of a prepared statement (asyncpg, psycopg after a few executions) cannot prune partitions by the query's parameters at plan time. The setting and the query run on one connection of the session: the one for the audit tables, that is their bind in Session(binds=...), else the session's default bind; with neither, UnboundExecutionError is raised.

FieldChange module-attribute

FieldChange: TypeAlias = (
    list[JSONValue] | RelationshipChange
)

A column change [old, new] or a relationship change.

ALL_HISTORY module-attribute

ALL_HISTORY = AllHistory.ALL_HISTORY

Pass as since to list the whole log, even with a default query window.

since=None means "not given", which applies the configured window.

FieldLabels module-attribute

FieldLabels: TypeAlias = (
    list[str | None] | RelationshipLabels
)

Labels of one change: [old, new] for a column (None where the value is None) or RelationshipLabels for a relationship.

ActivityData

Bases: _ActivityDataBase

The data envelope of an activity row.

Attributes:

Name	Type	Description
v	int	Envelope version.
changes	dict[str, FieldChange]	entity.* changes keyed by attribute name.
payload	dict[str, JSONValue]	Payload of a custom event.
context	ContextSnapshot	Request context snapshot taken when the entry was written.

ActivityRow

Bases: TypedDict

One audit_activity row, as fetched from the database.

Attributes:

Name	Type	Description
id	int	Row id.
transaction_id	int	Id of the audit transaction the row belongs to.
verb	str	Event verb, e.g. entity.updated.
severity	int	Severity value.
object_type	str | None	Type of the changed object.
object_id	str | None	Id of the changed object.
object_label	str | None	Label of the object when the row was written.
target_type	str | None	Type of the parent object.
target_id	str | None	Id of the parent object.
actor_id	str | None	Actor of the event.
scope_id	str | None	Scope, e.g. a tenant.
correlation_id	UUID | None	Correlation id of the transaction.
created_at	datetime	Start of the database transaction.
data	ActivityData	The data envelope.

AllHistory

Bases: Enum

Type of ALL_HISTORY: a listing without a lower time bound.

Cursor

Bases: NamedTuple

Position in the listing: (created_at, id) of an activity row.

list_groups continues with the rows after it in (created_at DESC, id DESC) order. The cursor also carries the lower time bound of the listing, so that every page uses the bound of the first one.

Attributes:

Name	Type	Description
created_at	datetime	created_at of the row; must be timezone-aware.
id	int	id of the row.
since	datetime | AllHistory | None	The listing's lower bound: a timezone-aware datetime, or ALL_HISTORY for none. A next_cursor always records it; None means not recorded (a cursor built by hand or a token from an older version), and the next page then resolves the bound as a first page would.

encode

encode() -> str

Return the cursor as an opaque, URL-safe token.

Returns:

Type	Description
str	Unpadded URL-safe base64 of <ISO created_at>|<id>, followed
str	by |<ISO since> or |* (ALL_HISTORY) when since is
str	recorded.

Source code in audit_trail/query.py

def encode(self) -> str:
    """Return the cursor as an opaque, URL-safe token.

    Returns:
        Unpadded URL-safe base64 of ``<ISO created_at>|<id>``, followed
        by ``|<ISO since>`` or ``|*`` (``ALL_HISTORY``) when ``since`` is
        recorded.
    """
    parts = [self.created_at.isoformat(), str(self.id)]
    match self.since:
        case None:
            pass
        case AllHistory.ALL_HISTORY:
            parts.append(_ALL_HISTORY_TOKEN)
        case datetime():
            parts.append(self.since.isoformat())
        case _:
            assert_never(self.since)
    raw = "|".join(parts).encode()
    return base64.urlsafe_b64encode(raw).decode().rstrip("=")

decode classmethod

decode(token: str) -> Cursor

Parse a token made by encode.

Parameters:

Name	Type	Description	Default
token	str	The token.	required

Returns:

Type	Description
Cursor	The cursor.

Raises:

Type	Description
ValueError	The token is malformed or one of its datetimes has no offset.

Source code in audit_trail/query.py

@classmethod
def decode(cls, token: str) -> Cursor:
    """Parse a token made by ``encode``.

    Args:
        token: The token.

    Returns:
        The cursor.

    Raises:
        ValueError: The token is malformed or one of its datetimes has no
            offset.
    """
    try:
        raw = base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))
        created_at, row_id, *rest = raw.decode().split("|")
        since: datetime | AllHistory | None
        match rest:
            case []:
                since = None
            case [str() as bound] if bound == _ALL_HISTORY_TOKEN:
                since = ALL_HISTORY
            case [str() as bound]:
                since = datetime.fromisoformat(bound)
            case _:
                raise ValueError("too many fields")
        cursor = cls(datetime.fromisoformat(created_at), int(row_id), since)
    except (binascii.Error, UnicodeDecodeError, ValueError) as exc:
        raise ValueError(f"malformed cursor: {token!r}") from exc
    _check_cursor(cursor)
    return cursor

Visibility dataclass

Visibility(
    object_types: Collection[str] | None = None,
    verbs: Collection[str] | None = None,
    scope_ids: Collection[str] | None = None,
)

What the caller may see; applied to every row the query returns.

The library knows nothing about the host's permissions: the host builds a Visibility from its own rules. It restricts both which transactions are listed and which of their entries a group shows.

Each field is a set of allowed values. None means no restriction; an empty collection allows nothing. A row whose column is NULL never passes a restriction that is set, so access control fails closed: with scope_ids={"t1"}, entries without a scope are hidden.

Attributes:

Name	Type	Description
object_types	Collection[str] | None	Allowed object_type values.
verbs	Collection[str] | None	Allowed verb values.
scope_ids	Collection[str] | None	Allowed scope_id values.

Raises:

Type	Description
TypeError	A field is a str instead of a collection of them.

predicate

predicate(table: Table) -> ColumnElement[bool]

Return the SQL condition on audit_activity rows.

Parameters:

Name	Type	Description	Default
table	Table	The audit_activity table.	required

Returns:

Type	Description
ColumnElement[bool]	The condition; true() when nothing is restricted.

Source code in audit_trail/query.py

def predicate(self, table: Table) -> ColumnElement[bool]:
    """Return the SQL condition on ``audit_activity`` rows.

    Args:
        table: The ``audit_activity`` table.

    Returns:
        The condition; ``true()`` when nothing is restricted.
    """
    conditions = [
        condition
        for condition in (
            _member_of(table.c.object_type, self.object_types),
            _member_of(table.c.verb, self.verbs),
            _member_of(table.c.scope_id, self.scope_ids),
        )
        if condition is not None
    ]
    return and_(*conditions) if conditions else true()

TransactionHeader dataclass

TransactionHeader(
    id: int,
    issued_at: datetime,
    actor_type: str | None,
    actor_id: str | None,
    actor_label: str | None,
    remote_addr: str | None,
    user_agent: str | None,
    method: str | None,
    path: str | None,
    channel: str | None,
    auth_method: str | None,
    request_id: UUID | None,
    correlation_id: UUID | None,
    scope_id: str | None,
    meta: dict[str, JSONValue] | None,
    from_snapshot: bool,
)

Who, where and how of one group: its audit_transaction row.

When the row no longer exists (retention), the header is built from the data.context snapshot of the group's earliest visible entry and from that entry's columns, and from_snapshot is True.

Attributes:

Name	Type	Description
id	int	The transaction id.
issued_at	datetime	Start of the database transaction (created_at of its entries).
actor_type	str | None	Kind of actor; None when a snapshot does not carry it.
actor_id	str | None	Actor id; from a snapshot, the entry's actor_id.
actor_label	str | None	Actor label when the transaction started.
remote_addr	str | None	Client address.
user_agent	str | None	Client user agent.
method	str | None	Request method.
path	str | None	Request path.
channel	str | None	Channel, e.g. api.
auth_method	str | None	Authentication method.
request_id	UUID | None	Request id.
correlation_id	UUID | None	Correlation id.
scope_id	str | None	scope_id of the group's earliest visible entry.
meta	dict[str, JSONValue] | None	Extra host context.
from_snapshot	bool	Whether the header was rebuilt from an entry.

Group dataclass

Group(
    transaction: TransactionHeader,
    activities: list[ActivityRow],
)

The entries of one database transaction.

Attributes:

Name	Type	Description
transaction	TransactionHeader	The transaction header.
activities	list[ActivityRow]	The visible entries of every severity, in id order, compacted unless the listing asked for raw rows. Never empty.

change_count property

change_count: int

Number of entries in the group.

max_severity property

max_severity: int

Highest severity among the group's entries.

Page dataclass

Page(
    groups: list[Group],
    next_cursor: Cursor | None,
    since: datetime | None = None,
)

One page of groups.

A page may hold fewer groups than the limit even when more follow (a group whose entries compaction hides entirely is left out); only next_cursor is None ends the listing.

Attributes:

Name	Type	Description
groups	list[Group]	Groups, newest first.
next_cursor	Cursor | None	Cursor of the next page, or None after the last one.
since	datetime | None	The lower time bound the listing used (created_at >= since), or None when it had none. With a default query window this is where the window starts: pass it as until to load the older entries.

ActivityDetail dataclass

ActivityDetail(
    activity: ActivityRow, transaction: TransactionHeader
)

One entry with the header of its transaction.

Attributes:

Name	Type	Description
activity	ActivityRow	The entry, as stored (not compacted).
transaction	TransactionHeader	The header of its transaction, rebuilt from the entry's snapshot when the audit_transaction row no longer exists.

AccessCount

Bases: NamedTuple

How often one actor accessed an object.

Attributes:

Name	Type	Description
actor_id	str | None	The actor; None for entries without an actor.
entries	int	Number of matching entries.
first_at	datetime	created_at of the actor's oldest matching entry.
last_at	datetime	created_at of the actor's newest matching entry.

RelationshipLabels

Bases: TypedDict

Labels of the members a relationship change added and removed.

Attributes:

Name	Type	Description
added	list[str]	Labels of the added members, in the order of their ids.
removed	list[str]	Labels of the removed members, in the order of their ids.

AuditQuery

AuditQuery(
    tables: AuditTables,
    severities: Iterable[int],
    *,
    default_window: timedelta | None = None,
)

Queries over the audit tables.

Parameters:

Name	Type	Description	Default
tables	AuditTables	The audit tables.	required
severities	Iterable[int]	Every severity in use; list_groups queries these when it is not given severities.	required
default_window	timedelta | None	How far back list_groups reads when it is not given since; None reads the whole log.	None

Raises:

Type	Description
TypeError	default_window is not a timedelta.
ValueError	default_window is not positive.

Source code in audit_trail/query.py

def __init__(
    self,
    tables: AuditTables,
    severities: Iterable[int],
    *,
    default_window: timedelta | None = None,
) -> None:
    self.tables = tables
    self.severities = tuple(sorted({int(severity) for severity in severities}))
    self.default_window = _check_window(default_window)

list_groups

list_groups(
    session: Session,
    *,
    severities: Collection[int] | None = None,
    since: datetime | AllHistory | None = None,
    until: datetime | None = None,
    actor_id: str | None = None,
    object_type: str | None = None,
    object_id: str | None = None,
    target_type: str | None = None,
    target_id: str | None = None,
    verbs: Collection[str] | None = None,
    scope_ids: Collection[str] | None = None,
    correlation_id: UUID | None = None,
    visibility: Visibility | None = None,
    extra_predicate: ColumnElement[bool] | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

List entries grouped by database transaction, newest first.

The filters select which transactions are listed: a transaction is listed when at least one of its entries matches all of them and visibility. A group then shows every entry of the transaction that visibility allows, of any severity: severities and the other filters narrow the list, they do not control access.

Collection filters take the allowed values: None means no restriction, an empty collection matches nothing. A NULL column never matches a restriction that is set.

Pagination counts transactions. A transaction appears whole on the page holding its newest matching entry and on no other, so following next_cursor returns every matching transaction exactly once, also when several transactions share a created_at.

With a default query window (AuditTrail(default_query_window=...)) and no since, the listing starts at until (or now) minus the window. The first page fixes that bound and its next_cursor carries it, so later pages neither move nor drop it. Page.since reports the bound: pass it as until to read the window before.

Statements run in the session's current transaction (one is begun if needed), on its connection for the audit tables, after SET LOCAL plan_cache_mode = 'force_custom_plan', which stays in effect until that transaction ends. On an AUTOCOMMIT connection the setting has no effect, so prepared statements may fall back to generic plans that read every partition.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
severities	Collection[int] | None	Severities to list; None lists all configured.	None
since	datetime | AllHistory | None	Only entries with created_at >= since. ALL_HISTORY sets no lower bound, even with a default window. None takes the bound recorded in cursor, else applies the default window, else sets no bound.	None
until	datetime | None	Only entries with created_at < until; also the end of the default window.	None
actor_id	str | None	Only entries of this actor.	None
object_type	str | None	Only entries on objects of this type.	None
object_id	str | None	Only entries on the object with this id.	None
target_type	str | None	Only entries whose parent object has this type.	None
target_id	str | None	Only entries whose parent object has this id.	None
verbs	Collection[str] | None	Only entries with one of these verbs.	None
scope_ids	Collection[str] | None	Only entries in one of these scopes.	None
correlation_id	UUID | None	Only entries with this correlation id.	None
visibility	Visibility | None	What the caller may see; None shows everything.	None
extra_predicate	ColumnElement[bool] | None	Further condition on audit_activity columns for selecting transactions; its performance is the host's concern.	None
cursor	Cursor | None	next_cursor of the previous page; None starts at the newest entry.	None
limit	int	Maximum number of groups on the page.	50
compact	bool	Merge each object's rows within a transaction (see compact_rows); False returns the raw rows.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	limit is below 1, or since, until or a datetime of the cursor is not timezone-aware.
TypeError	A collection filter is a str, or since is neither a datetime, ALL_HISTORY nor None.
UnboundExecutionError	The session has neither a bind for the audit tables nor a default bind.

Source code in audit_trail/query.py

def list_groups(
    self,
    session: Session,
    *,
    severities: Collection[int] | None = None,
    since: datetime | AllHistory | None = None,
    until: datetime | None = None,
    actor_id: str | None = None,
    object_type: str | None = None,
    object_id: str | None = None,
    target_type: str | None = None,
    target_id: str | None = None,
    verbs: Collection[str] | None = None,
    scope_ids: Collection[str] | None = None,
    correlation_id: UUID | None = None,
    visibility: Visibility | None = None,
    extra_predicate: ColumnElement[bool] | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """List entries grouped by database transaction, newest first.

    The filters select which transactions are listed: a transaction is
    listed when at least one of its entries matches all of them and
    ``visibility``. A group then shows every entry of the transaction
    that ``visibility`` allows, of any severity: ``severities`` and the
    other filters narrow the list, they do not control access.

    Collection filters take the allowed values: ``None`` means no
    restriction, an empty collection matches nothing. A ``NULL`` column
    never matches a restriction that is set.

    Pagination counts transactions. A transaction appears whole on the
    page holding its newest matching entry and on no other, so following
    ``next_cursor`` returns every matching transaction exactly once, also
    when several transactions share a ``created_at``.

    With a default query window (``AuditTrail(default_query_window=...)``)
    and no ``since``, the listing starts at ``until`` (or now) minus the
    window. The first page fixes that bound and its ``next_cursor``
    carries it, so later pages neither move nor drop it. ``Page.since``
    reports the bound: pass it as ``until`` to read the window before.

    Statements run in the session's current transaction (one is begun if
    needed), on its connection for the audit tables, after
    ``SET LOCAL plan_cache_mode = 'force_custom_plan'``, which stays in
    effect until that transaction ends. On an ``AUTOCOMMIT`` connection
    the setting has no effect, so prepared statements may fall back to
    generic plans that read every partition.

    Args:
        session: The session to query with.
        severities: Severities to list; ``None`` lists all configured.
        since: Only entries with ``created_at >= since``. ``ALL_HISTORY``
            sets no lower bound, even with a default window. ``None``
            takes the bound recorded in ``cursor``, else applies the
            default window, else sets no bound.
        until: Only entries with ``created_at < until``; also the end of
            the default window.
        actor_id: Only entries of this actor.
        object_type: Only entries on objects of this type.
        object_id: Only entries on the object with this id.
        target_type: Only entries whose parent object has this type.
        target_id: Only entries whose parent object has this id.
        verbs: Only entries with one of these verbs.
        scope_ids: Only entries in one of these scopes.
        correlation_id: Only entries with this correlation id.
        visibility: What the caller may see; ``None`` shows everything.
        extra_predicate: Further condition on ``audit_activity`` columns
            for selecting transactions; its performance is the host's
            concern.
        cursor: ``next_cursor`` of the previous page; ``None`` starts at
            the newest entry.
        limit: Maximum number of groups on the page.
        compact: Merge each object's rows within a transaction (see
            ``compact_rows``); ``False`` returns the raw rows.

    Returns:
        The page.

    Raises:
        ValueError: ``limit`` is below 1, or ``since``, ``until`` or a
            datetime of the cursor is not timezone-aware.
        TypeError: A collection filter is a ``str``, or ``since`` is
            neither a datetime, ``ALL_HISTORY`` nor ``None``.
        UnboundExecutionError: The session has neither a bind for the
            audit tables nor a default bind.
    """
    _check_page(since, until, cursor, limit)
    bound = self._bound(since, until, cursor, self.default_window)
    _check_collection(severities, "severities")
    _check_collection(verbs, "verbs")
    _check_collection(scope_ids, "scope_ids")

    a = self.tables.activity
    common: list[ColumnElement[bool]] = []
    equal: list[tuple[ColumnClause[object], object]] = [
        (a.c.actor_id, actor_id),
        (a.c.object_type, object_type),
        (a.c.object_id, object_id),
        (a.c.target_type, target_type),
        (a.c.target_id, target_id),
        (a.c.correlation_id, correlation_id),
    ]
    common.extend(field == value for field, value in equal if value is not None)
    for field, allowed in ((a.c.verb, verbs), (a.c.scope_id, scope_ids)):
        condition = _member_of(field, allowed)
        if condition is not None:
            common.append(condition)
    common.extend(self._window(_lower(bound), until, visibility))
    if extra_predicate is not None:
        common.append(extra_predicate)

    # An actor or scope filter without severities uses its own index in a
    # single query; otherwise one query per severity (see the module doc).
    streams: list[list[ColumnElement[bool]]]
    if severities is None and (actor_id is not None or scope_ids is not None):
        streams = [common]
        boundary = common
    else:
        wanted = self.severities if severities is None else sorted(set(severities))
        if not wanted:
            return Page(groups=[], next_cursor=None, since=_lower(bound))
        streams = [[a.c.severity == severity, *common] for severity in wanted]
        boundary = [a.c.severity.in_(wanted), *common]

    return self._page(
        session, streams, boundary, cursor, limit, visibility, compact, bound
    )

alist_groups async

alist_groups(
    session: AsyncSession,
    *,
    severities: Collection[int] | None = None,
    since: datetime | AllHistory | None = None,
    until: datetime | None = None,
    actor_id: str | None = None,
    object_type: str | None = None,
    object_id: str | None = None,
    target_type: str | None = None,
    target_id: str | None = None,
    verbs: Collection[str] | None = None,
    scope_ids: Collection[str] | None = None,
    correlation_id: UUID | None = None,
    visibility: Visibility | None = None,
    extra_predicate: ColumnElement[bool] | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

Async variant of list_groups, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
severities	Collection[int] | None	See list_groups.	None
since	datetime | AllHistory | None	See list_groups.	None
until	datetime | None	See list_groups.	None
actor_id	str | None	See list_groups.	None
object_type	str | None	See list_groups.	None
object_id	str | None	See list_groups.	None
target_type	str | None	See list_groups.	None
target_id	str | None	See list_groups.	None
verbs	Collection[str] | None	See list_groups.	None
scope_ids	Collection[str] | None	See list_groups.	None
correlation_id	UUID | None	See list_groups.	None
visibility	Visibility | None	See list_groups.	None
extra_predicate	ColumnElement[bool] | None	See list_groups.	None
cursor	Cursor | None	See list_groups.	None
limit	int	See list_groups.	50
compact	bool	See list_groups.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	See list_groups.
TypeError	See list_groups.

Source code in audit_trail/query.py

async def alist_groups(
    self,
    session: AsyncSession,
    *,
    severities: Collection[int] | None = None,
    since: datetime | AllHistory | None = None,
    until: datetime | None = None,
    actor_id: str | None = None,
    object_type: str | None = None,
    object_id: str | None = None,
    target_type: str | None = None,
    target_id: str | None = None,
    verbs: Collection[str] | None = None,
    scope_ids: Collection[str] | None = None,
    correlation_id: UUID | None = None,
    visibility: Visibility | None = None,
    extra_predicate: ColumnElement[bool] | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """Async variant of ``list_groups``, with the same arguments.

    Args:
        session: The async session to query with.
        severities: See ``list_groups``.
        since: See ``list_groups``.
        until: See ``list_groups``.
        actor_id: See ``list_groups``.
        object_type: See ``list_groups``.
        object_id: See ``list_groups``.
        target_type: See ``list_groups``.
        target_id: See ``list_groups``.
        verbs: See ``list_groups``.
        scope_ids: See ``list_groups``.
        correlation_id: See ``list_groups``.
        visibility: See ``list_groups``.
        extra_predicate: See ``list_groups``.
        cursor: See ``list_groups``.
        limit: See ``list_groups``.
        compact: See ``list_groups``.

    Returns:
        The page.

    Raises:
        ValueError: See ``list_groups``.
        TypeError: See ``list_groups``.
    """
    return await session.run_sync(
        lambda sync_session: self.list_groups(
            sync_session,
            severities=severities,
            since=since,
            until=until,
            actor_id=actor_id,
            object_type=object_type,
            object_id=object_id,
            target_type=target_type,
            target_id=target_id,
            verbs=verbs,
            scope_ids=scope_ids,
            correlation_id=correlation_id,
            visibility=visibility,
            extra_predicate=extra_predicate,
            cursor=cursor,
            limit=limit,
            compact=compact,
        )
    )

get

get(
    session: Session,
    id: int,
    created_at: datetime,
    *,
    severity: int | None = None,
    visibility: Visibility | None = None,
) -> ActivityDetail | None

Return one entry with its transaction header.

An entry is found by id and created_at together: id alone would read every partition, while created_at limits the read to one monthly partition of each severity, and severity (the entry's own, as a listing row carries it) to exactly one partition.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
id	int	The entry's id.	required
created_at	datetime	The entry's created_at.	required
severity	int | None	The entry's severity, if known.	None
visibility	Visibility | None	What the caller may see; None shows everything.	None

Returns:

Type	Description
ActivityDetail | None	The entry and its header, or None when there is no such entry
ActivityDetail | None	or visibility hides it; the two cases are indistinguishable.

Raises:

Type	Description
ValueError	created_at is not timezone-aware.

Source code in audit_trail/query.py

def get(
    self,
    session: Session,
    id: int,
    created_at: datetime,
    *,
    severity: int | None = None,
    visibility: Visibility | None = None,
) -> ActivityDetail | None:
    """Return one entry with its transaction header.

    An entry is found by ``id`` and ``created_at`` together: ``id`` alone
    would read every partition, while ``created_at`` limits the read to
    one monthly partition of each severity, and ``severity`` (the entry's
    own, as a listing row carries it) to exactly one partition.

    Args:
        session: The session to query with.
        id: The entry's ``id``.
        created_at: The entry's ``created_at``.
        severity: The entry's severity, if known.
        visibility: What the caller may see; ``None`` shows everything.

    Returns:
        The entry and its header, or ``None`` when there is no such entry
        or ``visibility`` hides it; the two cases are indistinguishable.

    Raises:
        ValueError: ``created_at`` is not timezone-aware.
    """
    _check_aware(created_at, "created_at")
    a = self.tables.activity
    t = self.tables.transaction
    conditions = [a.c.id == id, a.c.created_at == created_at]
    if severity is not None:
        conditions.append(a.c.severity == severity)
    if visibility is not None:
        conditions.append(visibility.predicate(a))
    connection = audit_connection(session, self.tables)
    connection.execute(_CUSTOM_PLAN)
    row = connection.execute(select(a).where(*conditions)).mappings().first()
    if row is None:
        return None
    activity = _activity(row)
    header = (
        connection.execute(
            select(t).where(
                t.c.id == activity["transaction_id"],
                t.c.issued_at == activity["created_at"],
            )
        )
        .mappings()
        .first()
    )
    return ActivityDetail(activity=activity, transaction=_header(header, activity))

aget async

aget(
    session: AsyncSession,
    id: int,
    created_at: datetime,
    *,
    severity: int | None = None,
    visibility: Visibility | None = None,
) -> ActivityDetail | None

Async variant of get, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
id	int	See get.	required
created_at	datetime	See get.	required
severity	int | None	See get.	None
visibility	Visibility | None	See get.	None

Returns:

Type	Description
ActivityDetail | None	See get.

Raises:

Type	Description
ValueError	See get.

Source code in audit_trail/query.py

async def aget(
    self,
    session: AsyncSession,
    id: int,
    created_at: datetime,
    *,
    severity: int | None = None,
    visibility: Visibility | None = None,
) -> ActivityDetail | None:
    """Async variant of ``get``, with the same arguments.

    Args:
        session: The async session to query with.
        id: See ``get``.
        created_at: See ``get``.
        severity: See ``get``.
        visibility: See ``get``.

    Returns:
        See ``get``.

    Raises:
        ValueError: See ``get``.
    """
    return await session.run_sync(
        lambda sync_session: self.get(
            sync_session,
            id,
            created_at,
            severity=severity,
            visibility=visibility,
        )
    )

object_history

object_history(
    session: Session,
    object_type: str,
    object_id: str,
    *,
    include_children: bool = True,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

List the history of one record, grouped by database transaction.

The history holds the entries on the record and, with include_children, the entries whose target is the record (the changes of its children). Each group shows only those entries of its transaction, not the transaction's other entries. visibility applies to every entry on its own: children of types the caller may not see are left out, and a visible child's entries are listed even when the record's own type is hidden.

A child's entries follow the target it had when they were written: a child moved to another parent appears, from the move on, in the new parent's history only.

Pagination works as in list_groups: it counts transactions, newest first, and next_cursor returns each transaction once. The default query window does not apply: without since (given or recorded in cursor) the whole history is listed.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
object_type	str	The record's object_type.	required
object_id	str	The record's object_id, formatted as object_id_for() formats it; any other string matches nothing.	required
include_children	bool	Also list the entries targeting the record.	True
since	datetime | None	Only entries with created_at >= since.	None
until	datetime | None	Only entries with created_at < until.	None
visibility	Visibility | None	What the caller may see; None shows everything.	None
cursor	Cursor | None	next_cursor of the previous page; None starts at the newest entry.	None
limit	int	Maximum number of groups on the page.	50
compact	bool	Merge each object's rows within a transaction; False returns the raw rows.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	limit is below 1, or since, until or the cursor's created_at is not timezone-aware.

Source code in audit_trail/query.py

def object_history(
    self,
    session: Session,
    object_type: str,
    object_id: str,
    *,
    include_children: bool = True,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """List the history of one record, grouped by database transaction.

    The history holds the entries on the record and, with
    ``include_children``, the entries whose target is the record (the
    changes of its children). Each group shows only those entries of its
    transaction, not the transaction's other entries. ``visibility``
    applies to every entry on its own: children of types the caller may
    not see are left out, and a visible child's entries are listed even
    when the record's own type is hidden.

    A child's entries follow the target it had when they were written: a
    child moved to another parent appears, from the move on, in the new
    parent's history only.

    Pagination works as in ``list_groups``: it counts transactions,
    newest first, and ``next_cursor`` returns each transaction once. The
    default query window does not apply: without ``since`` (given or
    recorded in ``cursor``) the whole history is listed.

    Args:
        session: The session to query with.
        object_type: The record's ``object_type``.
        object_id: The record's ``object_id``, formatted as
            ``object_id_for()`` formats it; any other string matches
            nothing.
        include_children: Also list the entries targeting the record.
        since: Only entries with ``created_at >= since``.
        until: Only entries with ``created_at < until``.
        visibility: What the caller may see; ``None`` shows everything.
        cursor: ``next_cursor`` of the previous page; ``None`` starts at
            the newest entry.
        limit: Maximum number of groups on the page.
        compact: Merge each object's rows within a transaction;
            ``False`` returns the raw rows.

    Returns:
        The page.

    Raises:
        ValueError: ``limit`` is below 1, or ``since``, ``until`` or the
            cursor's ``created_at`` is not timezone-aware.
    """
    _check_page(since, until, cursor, limit)
    bound = self._bound(since, until, cursor, None)
    a = self.tables.activity
    common = self._window(_lower(bound), until, visibility)
    own = and_(a.c.object_type == object_type, a.c.object_id == object_id)
    # Two streams, one per index, merged in Python: an OR of both
    # conditions could not read either index in order.
    streams = [[own, *common]]
    scope: ColumnElement[bool] = own
    if include_children:
        children = and_(a.c.target_type == object_type, a.c.target_id == object_id)
        streams.append([children, *common])
        scope = or_(own, children)
    return self._page(
        session,
        streams,
        [scope, *common],
        cursor,
        limit,
        visibility,
        compact,
        bound,
        scope=scope,
    )

aobject_history async

aobject_history(
    session: AsyncSession,
    object_type: str,
    object_id: str,
    *,
    include_children: bool = True,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

Async variant of object_history, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
object_type	str	See object_history.	required
object_id	str	See object_history.	required
include_children	bool	See object_history.	True
since	datetime | None	See object_history.	None
until	datetime | None	See object_history.	None
visibility	Visibility | None	See object_history.	None
cursor	Cursor | None	See object_history.	None
limit	int	See object_history.	50
compact	bool	See object_history.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	See object_history.

Source code in audit_trail/query.py

async def aobject_history(
    self,
    session: AsyncSession,
    object_type: str,
    object_id: str,
    *,
    include_children: bool = True,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """Async variant of ``object_history``, with the same arguments.

    Args:
        session: The async session to query with.
        object_type: See ``object_history``.
        object_id: See ``object_history``.
        include_children: See ``object_history``.
        since: See ``object_history``.
        until: See ``object_history``.
        visibility: See ``object_history``.
        cursor: See ``object_history``.
        limit: See ``object_history``.
        compact: See ``object_history``.

    Returns:
        The page.

    Raises:
        ValueError: See ``object_history``.
    """
    return await session.run_sync(
        lambda sync_session: self.object_history(
            sync_session,
            object_type,
            object_id,
            include_children=include_children,
            since=since,
            until=until,
            visibility=visibility,
            cursor=cursor,
            limit=limit,
            compact=compact,
        )
    )

related

related(
    session: Session,
    correlation_id: UUID,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

List the transactions sharing a correlation id, newest first.

Groups and pagination are those of list_groups; the transactions are selected with one query over the correlation index. The default query window does not apply, as for object_history.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
correlation_id	UUID	The correlation id.	required
since	datetime | None	Only entries with created_at >= since.	None
until	datetime | None	Only entries with created_at < until.	None
visibility	Visibility | None	What the caller may see; None shows everything.	None
cursor	Cursor | None	next_cursor of the previous page; None starts at the newest entry.	None
limit	int	Maximum number of groups on the page.	50
compact	bool	Merge each object's rows within a transaction; False returns the raw rows.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	limit is below 1, or since, until or the cursor's created_at is not timezone-aware.

Source code in audit_trail/query.py

def related(
    self,
    session: Session,
    correlation_id: UUID,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """List the transactions sharing a correlation id, newest first.

    Groups and pagination are those of ``list_groups``; the transactions
    are selected with one query over the correlation index. The default
    query window does not apply, as for ``object_history``.

    Args:
        session: The session to query with.
        correlation_id: The correlation id.
        since: Only entries with ``created_at >= since``.
        until: Only entries with ``created_at < until``.
        visibility: What the caller may see; ``None`` shows everything.
        cursor: ``next_cursor`` of the previous page; ``None`` starts at
            the newest entry.
        limit: Maximum number of groups on the page.
        compact: Merge each object's rows within a transaction;
            ``False`` returns the raw rows.

    Returns:
        The page.

    Raises:
        ValueError: ``limit`` is below 1, or ``since``, ``until`` or the
            cursor's ``created_at`` is not timezone-aware.
    """
    _check_page(since, until, cursor, limit)
    bound = self._bound(since, until, cursor, None)
    a = self.tables.activity
    conditions = [
        a.c.correlation_id == correlation_id,
        *self._window(_lower(bound), until, visibility),
    ]
    return self._page(
        session, [conditions], conditions, cursor, limit, visibility, compact, bound
    )

arelated async

arelated(
    session: AsyncSession,
    correlation_id: UUID,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page

Async variant of related, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
correlation_id	UUID	See related.	required
since	datetime | None	See related.	None
until	datetime | None	See related.	None
visibility	Visibility | None	See related.	None
cursor	Cursor | None	See related.	None
limit	int	See related.	50
compact	bool	See related.	True

Returns:

Type	Description
Page	The page.

Raises:

Type	Description
ValueError	See related.

Source code in audit_trail/query.py

async def arelated(
    self,
    session: AsyncSession,
    correlation_id: UUID,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
    cursor: Cursor | None = None,
    limit: int = 50,
    compact: bool = True,
) -> Page:
    """Async variant of ``related``, with the same arguments.

    Args:
        session: The async session to query with.
        correlation_id: See ``related``.
        since: See ``related``.
        until: See ``related``.
        visibility: See ``related``.
        cursor: See ``related``.
        limit: See ``related``.
        compact: See ``related``.

    Returns:
        The page.

    Raises:
        ValueError: See ``related``.
    """
    return await session.run_sync(
        lambda sync_session: self.related(
            sync_session,
            correlation_id,
            since=since,
            until=until,
            visibility=visibility,
            cursor=cursor,
            limit=limit,
            compact=compact,
        )
    )

access_summary

access_summary(
    session: Session,
    object_type: str,
    object_id: str,
    verb: str,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
) -> list[AccessCount]

Count one verb's entries on a record per actor.

For example who read a person's sensitive data, and how often. Entries are counted raw, without compaction. The default query window does not apply: without since every entry is counted.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
object_type	str	The record's object_type.	required
object_id	str	The record's object_id (see object_id_for).	required
verb	str	The verb to count, e.g. person.viewed.	required
since	datetime | None	Only entries with created_at >= since.	None
until	datetime | None	Only entries with created_at < until.	None
visibility	Visibility | None	What the caller may see; None counts everything.	None

Returns:

Type	Description
list[AccessCount]	One count per actor, the most recent access first.

Raises:

Type	Description
ValueError	since or until is not timezone-aware.

Source code in audit_trail/query.py

def access_summary(
    self,
    session: Session,
    object_type: str,
    object_id: str,
    verb: str,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
) -> list[AccessCount]:
    """Count one verb's entries on a record per actor.

    For example who read a person's sensitive data, and how often.
    Entries are counted raw, without compaction. The default query window
    does not apply: without ``since`` every entry is counted.

    Args:
        session: The session to query with.
        object_type: The record's ``object_type``.
        object_id: The record's ``object_id`` (see ``object_id_for``).
        verb: The verb to count, e.g. ``person.viewed``.
        since: Only entries with ``created_at >= since``.
        until: Only entries with ``created_at < until``.
        visibility: What the caller may see; ``None`` counts everything.

    Returns:
        One count per actor, the most recent access first.

    Raises:
        ValueError: ``since`` or ``until`` is not timezone-aware.
    """
    _check_aware(since, "since")
    _check_aware(until, "until")
    a = self.tables.activity
    last_at = func.max(a.c.created_at)
    statement = (
        select(a.c.actor_id, func.count(), func.min(a.c.created_at), last_at)
        .where(
            a.c.object_type == object_type,
            a.c.object_id == object_id,
            a.c.verb == verb,
            *self._window(since, until, visibility),
        )
        .group_by(a.c.actor_id)
        .order_by(last_at.desc(), a.c.actor_id)
    )
    connection = audit_connection(session, self.tables)
    connection.execute(_CUSTOM_PLAN)
    return [AccessCount(*row) for row in connection.execute(statement)]

aaccess_summary async

aaccess_summary(
    session: AsyncSession,
    object_type: str,
    object_id: str,
    verb: str,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
) -> list[AccessCount]

Async variant of access_summary, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
object_type	str	See access_summary.	required
object_id	str	See access_summary.	required
verb	str	See access_summary.	required
since	datetime | None	See access_summary.	None
until	datetime | None	See access_summary.	None
visibility	Visibility | None	See access_summary.	None

Returns:

Type	Description
list[AccessCount]	See access_summary.

Raises:

Type	Description
ValueError	See access_summary.

Source code in audit_trail/query.py

async def aaccess_summary(
    self,
    session: AsyncSession,
    object_type: str,
    object_id: str,
    verb: str,
    *,
    since: datetime | None = None,
    until: datetime | None = None,
    visibility: Visibility | None = None,
) -> list[AccessCount]:
    """Async variant of ``access_summary``, with the same arguments.

    Args:
        session: The async session to query with.
        object_type: See ``access_summary``.
        object_id: See ``access_summary``.
        verb: See ``access_summary``.
        since: See ``access_summary``.
        until: See ``access_summary``.
        visibility: See ``access_summary``.

    Returns:
        See ``access_summary``.

    Raises:
        ValueError: See ``access_summary``.
    """
    return await session.run_sync(
        lambda sync_session: self.access_summary(
            sync_session,
            object_type,
            object_id,
            verb,
            since=since,
            until=until,
            visibility=visibility,
        )
    )

LabelResolver

LabelResolver(
    tables: AuditTables, base: registry | type[Any]
)

Labels for the objects that entries' foreign keys and relationships name.

A change such as {"board_id": [1, 2]} stores ids. The resolver finds which fields refer to other objects from the models' metadata, never from field names: a column with a single-column ForeignKey to the primary key of a mapped table, and every relationship (whose tracked changes list member ids). It then looks each referenced object up in the audit trail, not in the host's tables.

The label of an object is the newest non-null object_label among its entries that visibility lets the caller see. So a deleted object keeps the label of its entity.deleted entry, and an object the caller may not see (its type, scope or verbs hidden) is not revealed: it gets the fallback #<id>, as does an object without a labelled entry, e.g. of a model without a label option. A label is as fresh as the object's last audited change: an unaudited rename is not reflected.

Not resolved: composite foreign keys, foreign keys to columns other than the primary key, fields with an audit policy (redact, hash) and redacted or unknown values, as well as target_type/target_id.

Parameters:

Name	Type	Description	Default
tables	AuditTables	The audit tables.	required
base	registry | type[Any]	The registry of the models, or a declarative base class using it, as for check_models. Its mappers are configured.	required

Source code in audit_trail/query.py

def __init__(self, tables: AuditTables, base: registry | type[Any]) -> None:
    self.tables = tables
    reg = _registry_of(base)
    reg.configure()
    mappers = list(reg.mappers)
    self._labelled = frozenset(
        _type_name(mapper.class_)
        for mapper in mappers
        if options_of(mapper.class_).label is not None
    )
    self._fields: dict[str, dict[str, frozenset[str]]] = {}
    for mapper in mappers:
        fields = self._fields.setdefault(_type_name(mapper.class_), {})
        for prop in mapper.column_attrs:
            types = _referenced_types(mapper, prop.key, prop.columns, mappers)
            if types:
                fields[prop.key] = types
        for relationship in mapper.relationships:
            fields[relationship.key] = _types_of(relationship.mapper)

resolve

resolve(
    session: Session,
    activities: Iterable[ActivityRow],
    *,
    visibility: Visibility | None = None,
) -> dict[int, dict[str, FieldLabels]]

Label the referenced objects of the entries' changes.

Parameters:

Name	Type	Description	Default
session	Session	The session to query with.	required
activities	Iterable[ActivityRow]	The entries, e.g. those of a page's groups.	required
visibility	Visibility | None	What the caller may see; None shows everything.	None

Returns:

Name	Type	Description
	dict[int, dict[str, FieldLabels]]	By entry id, then by field, the labels in the shape of the
change	dict[int, dict[str, FieldLabels]]	[old, new] for a column, RelationshipLabels for a
	dict[int, dict[str, FieldLabels]]	relationship. Entries without reference fields are left out.

Source code in audit_trail/query.py

def resolve(
    self,
    session: Session,
    activities: Iterable[ActivityRow],
    *,
    visibility: Visibility | None = None,
) -> dict[int, dict[str, FieldLabels]]:
    """Label the referenced objects of the entries' changes.

    Args:
        session: The session to query with.
        activities: The entries, e.g. those of a page's groups.
        visibility: What the caller may see; ``None`` shows everything.

    Returns:
        By entry ``id``, then by field, the labels in the shape of the
        change: ``[old, new]`` for a column, ``RelationshipLabels`` for a
        relationship. Entries without reference fields are left out.
    """
    found: list[_Found] = []
    wanted: set[_Reference] = set()
    for activity in activities:
        object_type = activity["object_type"]
        fields = {} if object_type is None else self._fields.get(object_type, {})
        for key, change in activity["data"].get("changes", {}).items():
            types = fields.get(key)
            ids = None if types is None else _referenced_ids(change)
            if types is None or ids is None:
                continue
            found.append(_Found(activity["id"], key, change, types))
            wanted.update(
                _Reference(object_type, object_id)
                for object_type in types & self._labelled
                for object_id in ids
            )
    labels = self._labels(session, wanted, visibility)

    def label(types: frozenset[str], object_id: str) -> str:
        hits = [
            labels[reference]
            for reference in (_Reference(t, object_id) for t in types)
            if reference in labels
        ]
        return max(hits).label if hits else f"#{object_id}"

    result: dict[int, dict[str, FieldLabels]] = {}
    for item in found:
        change = item.change
        labelled: FieldLabels
        if isinstance(change, list):
            labelled = [
                None if value is None else label(item.types, str(value))
                for value in change
            ]
        else:
            labelled = {
                "added": [label(item.types, i) for i in change["added"]],
                "removed": [label(item.types, i) for i in change["removed"]],
            }
        result.setdefault(item.activity_id, {})[item.key] = labelled
    return result

aresolve async

aresolve(
    session: AsyncSession,
    activities: Iterable[ActivityRow],
    *,
    visibility: Visibility | None = None,
) -> dict[int, dict[str, FieldLabels]]

Async variant of resolve, with the same arguments.

Parameters:

Name	Type	Description	Default
session	AsyncSession	The async session to query with.	required
activities	Iterable[ActivityRow]	See resolve.	required
visibility	Visibility | None	See resolve.	None

Returns:

Type	Description
dict[int, dict[str, FieldLabels]]	See resolve.

Source code in audit_trail/query.py

async def aresolve(
    self,
    session: AsyncSession,
    activities: Iterable[ActivityRow],
    *,
    visibility: Visibility | None = None,
) -> dict[int, dict[str, FieldLabels]]:
    """Async variant of ``resolve``, with the same arguments.

    Args:
        session: The async session to query with.
        activities: See ``resolve``.
        visibility: See ``resolve``.

    Returns:
        See ``resolve``.
    """
    rows = list(activities)
    return await session.run_sync(
        lambda sync_session: self.resolve(sync_session, rows, visibility=visibility)
    )

changed_fields

changed_fields(activity: ActivityRow) -> list[str]

Return the names of the fields an entry changed.

Parameters:

Name	Type	Description	Default
activity	ActivityRow	An entry.	required

Returns:

Type	Description
list[str]	The keys of data.changes; empty for entries without changes.

Source code in audit_trail/query.py

def changed_fields(activity: ActivityRow) -> list[str]:
    """Return the names of the fields an entry changed.

    Args:
        activity: An entry.

    Returns:
        The keys of ``data.changes``; empty for entries without changes.
    """
    return list(activity["data"].get("changes", {}))