audit_trail.tables

Core table definitions for the audit log.

Both tables are partitioned in PostgreSQL, so their primary keys include the partition keys and nothing has a foreign key to them:

• audit_transaction: PARTITION BY RANGE (issued_at), one partition per month.
• audit_activity: PARTITION BY LIST (severity), and each severity partition is itself PARTITION BY RANGE (created_at), one partition per month.

The partitions are not part of the metadata; audit_trail.migrations creates the parents and severity partitions, audit_trail.maintenance the monthly ones.

MAX_IDENTIFIER_LENGTH module-attribute

MAX_IDENTIFIER_LENGTH = 63

PostgreSQL's limit on identifier length in bytes; longer names are truncated.

IndexKey module-attribute

IndexKey = Literal[
    "severity",
    "actor",
    "object",
    "target",
    "scope",
    "transaction",
    "correlation",
    "changes_gin",
]

Key of one audit_activity index.

DEFAULT_INDEXES module-attribute

DEFAULT_INDEXES: frozenset[IndexKey] = frozenset(
    {
        "severity",
        "actor",
        "object",
        "target",
        "scope",
        "transaction",
        "correlation",
    }
)

Keys of the audit_activity indexes created unless indexes says otherwise.

OPTIONAL_INDEXES module-attribute

OPTIONAL_INDEXES: frozenset[IndexKey] = frozenset(
    {"changes_gin"}
)

Keys of the indexes that exist but are off by default.

ALL_INDEXES module-attribute

ALL_INDEXES: frozenset[IndexKey] = frozenset(
    get_args(IndexKey)
)

Every valid key for build_tables(indexes=...).

AuditTables dataclass

AuditTables(
    metadata: MetaData, transaction: Table, activity: Table
)

The audit tables and the metadata they are bound to.

Attributes:

Name	Type	Description
metadata	MetaData	Metadata holding both tables and the activity indexes.
transaction	Table	The audit_transaction table.
activity	Table	The audit_activity table.

build_tables

build_tables(
    *,
    schema: str = "audit",
    transaction_table: str = "audit_transaction",
    activity_table: str = "audit_activity",
    indexes: Collection[str] | None = None,
) -> AuditTables

Build the audit tables in a new MetaData.

Parameters:

Name	Type	Description	Default
schema	str	Schema holding the tables.	'audit'
transaction_table	str	Name of the transaction (context) table.	'audit_transaction'
activity_table	str	Name of the activity (event) table.	'audit_activity'
indexes	Collection[str] | None	Keys of the audit_activity indexes to create, from ALL_INDEXES. None uses DEFAULT_INDEXES. changes_gin is a GIN index on data -> 'changes' and is off by default.	None

Returns:

Type	Description
AuditTables	The tables and their metadata.

Raises:

Type	Description
ValueError	If a name is empty or too long for PostgreSQL once a partition suffix is added, the table names are equal, or an index key is unknown.

Source code in audit_trail/tables.py

def build_tables(
    *,
    schema: str = "audit",
    transaction_table: str = "audit_transaction",
    activity_table: str = "audit_activity",
    indexes: Collection[str] | None = None,
) -> AuditTables:
    """Build the audit tables in a new ``MetaData``.

    Args:
        schema: Schema holding the tables.
        transaction_table: Name of the transaction (context) table.
        activity_table: Name of the activity (event) table.
        indexes: Keys of the ``audit_activity`` indexes to create, from
            ``ALL_INDEXES``. ``None`` uses ``DEFAULT_INDEXES``. ``changes_gin``
            is a GIN index on ``data -> 'changes'`` and is off by default.

    Returns:
        The tables and their metadata.

    Raises:
        ValueError: If a name is empty or too long for PostgreSQL once a
            partition suffix is added, the table names are equal, or an index
            key is unknown.
    """
    _check_name(schema, "schema", 0)
    _check_name(transaction_table, "transaction_table", _PARTITION_SUFFIX_LENGTH)
    _check_name(activity_table, "activity_table", _PARTITION_SUFFIX_LENGTH)
    if transaction_table == activity_table:
        raise ValueError("transaction_table and activity_table must differ")
    if indexes is None:
        index_keys = DEFAULT_INDEXES
    else:
        unknown = frozenset(indexes) - ALL_INDEXES
        if unknown:
            raise ValueError(f"unknown index keys: {sorted(unknown)}")
        index_keys = ALL_INDEXES & frozenset(indexes)

    metadata = MetaData(schema=schema)
    transaction = Table(
        transaction_table,
        metadata,
        Column("id", BigInteger, Identity(always=True), nullable=False),
        Column(
            "issued_at",
            DateTime(timezone=True),
            nullable=False,
            server_default=text("now()"),
        ),
        Column("actor_type", Text, nullable=False),
        Column("actor_id", Text),
        Column("actor_label", Text),
        Column("remote_addr", INET),
        Column("user_agent", Text),
        Column("method", Text),
        Column("path", Text),
        Column("channel", Text),
        Column("auth_method", Text),
        Column("request_id", UUID(as_uuid=True)),
        Column("correlation_id", UUID(as_uuid=True)),
        Column("meta", JSONB),
        PrimaryKeyConstraint("id", "issued_at", name=f"{transaction_table}_pkey"),
        postgresql_partition_by="RANGE (issued_at)",
    )
    activity = Table(
        activity_table,
        metadata,
        Column("id", BigInteger, Identity(always=True), nullable=False),
        Column("transaction_id", BigInteger, nullable=False),
        Column("verb", Text, nullable=False),
        Column("severity", SmallInteger, nullable=False),
        Column("object_type", Text),
        Column("object_id", Text),
        Column("object_label", Text),
        Column("target_type", Text),
        Column("target_id", Text),
        Column("actor_id", Text),
        Column("scope_id", Text),
        Column("correlation_id", UUID(as_uuid=True)),
        Column("created_at", DateTime(timezone=True), nullable=False),
        Column("data", JSONB, nullable=False, server_default=text("'{}'::jsonb")),
        PrimaryKeyConstraint(
            "id", "severity", "created_at", name=f"{activity_table}_pkey"
        ),
        postgresql_partition_by="LIST (severity)",
    )
    for key in sorted(index_keys):
        _activity_index(activity, key)
    return AuditTables(metadata=metadata, transaction=transaction, activity=activity)