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.

Both stages run 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.

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.

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.

Attributes:

Name	Type	Description
created_at	datetime	created_at of the row; must be timezone-aware.
id	int	id of the row.

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>.

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>``.
    """
    raw = f"{self.created_at.isoformat()}|{self.id}".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 its datetime 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 its datetime has no offset.
    """
    try:
        raw = base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))
        created_at, row_id = raw.decode().split("|")
        cursor = cls(datetime.fromisoformat(created_at), int(row_id))
    except (binascii.Error, UnicodeDecodeError, ValueError) as exc:
        raise ValueError(f"malformed cursor: {token!r}") from exc
    _check_aware(cursor.created_at, "cursor created_at")
    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)

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.

AuditQuery

AuditQuery(tables: AuditTables, severities: Iterable[int])

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

Source code in audit_trail/query.py

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

list_groups

list_groups(
    session: Session,
    *,
    severities: Collection[int] | None = None,
    since: datetime | 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.

Statements run in the session's current transaction (one is begun if needed) 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 | None	Only entries with created_at >= since.	None
until	datetime | None	Only entries with created_at < until.	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 the cursor's created_at is not timezone-aware.
TypeError	A collection filter is a str.

Source code in audit_trail/query.py

def list_groups(
    self,
    session: Session,
    *,
    severities: Collection[int] | None = None,
    since: datetime | 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``.

    Statements run in the session's current transaction (one is begun if
    needed) 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``.
        until: Only entries with ``created_at < until``.
        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 the
            cursor's ``created_at`` is not timezone-aware.
        TypeError: A collection filter is a ``str``.
    """
    if limit < 1:
        raise ValueError("limit must be at least 1")
    _check_aware(since, "since")
    _check_aware(until, "until")
    if cursor is not None:
        _check_aware(cursor.created_at, "cursor created_at")
    _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(column == value for column, value in equal if value is not None)
    for column, values in ((a.c.verb, verbs), (a.c.scope_id, scope_ids)):
        condition = _member_of(column, values)
        if condition is not None:
            common.append(condition)
    if since is not None:
        common.append(a.c.created_at >= since)
    if until is not None:
        common.append(a.c.created_at < until)
    if visibility is not None:
        common.append(visibility.predicate(a))
    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)
        streams = [[a.c.severity == severity, *common] for severity in wanted]
        boundary = [a.c.severity.in_(wanted), *common]

    session.execute(_CUSTOM_PLAN)
    shown = self._shown(session, boundary, cursor)
    selection = self._select(session, streams, cursor, shown, limit)
    groups = self._groups(session, selection.transactions, visibility, compact)
    return Page(groups=groups, next_cursor=selection.next_cursor)

alist_groups async

alist_groups(
    session: AsyncSession,
    *,
    severities: Collection[int] | None = None,
    since: datetime | 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 | 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 | 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,
        )
    )

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", {}))