audit_trail.diff

Entity change sets and per-column field policies.

Everything here reads the in-memory state of ORM instances and never emits SQL, so it is safe to call from after_flush (also under AsyncSession).

A column's policy is set where the column is defined::

hook_secret: Mapped[str] = mapped_column(info={"audit": "redact"})
email: Mapped[str] = mapped_column(info={"audit": "hash"})
embeddings: Mapped[bytes] = mapped_column(info={"audit": "exclude"})

FieldPolicy module-attribute

FieldPolicy: TypeAlias = Literal[
    "exclude", "redact", "hash"
]

Policy of one column, from mapped_column(info={"audit": ...}).

ChangeKind module-attribute

ChangeKind: TypeAlias = Literal[
    "created", "updated", "deleted"
]

Which change set to build: a new, a modified or a deleted instance.

Changes module-attribute

Changes: TypeAlias = dict[str, list[JSONValue]]

{attribute key: [old, new]} with JSON-ready values.

REDACTED module-attribute

REDACTED = '***'

Stored in place of a non-null value of a redact column.

UNKNOWN module-attribute

UNKNOWN = '<unknown>'

Marker stored where a value is not available without SQL.

ERASED module-attribute

ERASED = '[erased]'

Stored by scrub in place of a non-null value.

SNAPSHOT_INFO_KEY module-attribute

SNAPSHOT_INFO_KEY = 'audit_trail.snapshot'

InstanceState.info key holding the snapshot_on_load copies.

FieldPolicyError

Bases: ValueError

A column's audit policy is invalid or cannot be applied.

AuditOptionError

Bases: TypeError

An audit option (label, scope, target) is misconfigured.

UseContext

Bases: Enum

Result of :func:resolve_scope meaning "take the scope from context".

ResolvedTarget

Bases: NamedTuple

A target as stored in target_type and target_id.

id instance-attribute

id: str

Formatted like an object_id.

options_of

options_of(model: type[Any]) -> AuditOptions

Return the model's __audit__ options, or the defaults.

Parameters:

Name	Type	Description	Default
model	type[Any]	A mapped class.	required

Returns:

Type	Description
AuditOptions	The model's AuditOptions.

Source code in audit_trail/diff.py

def options_of(model: type[Any]) -> AuditOptions:
    """Return the model's ``__audit__`` options, or the defaults.

    Args:
        model: A mapped class.

    Returns:
        The model's ``AuditOptions``.
    """
    options = getattr(model, "__audit__", None)
    return options if isinstance(options, AuditOptions) else AuditOptions()

object_type_of

object_type_of(obj: object) -> str

Return the object_type of an instance.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required

Returns:

Type	Description
str	AuditOptions.object_type, or the class name when it is not set.

Source code in audit_trail/diff.py

def object_type_of(obj: object) -> str:
    """Return the ``object_type`` of an instance.

    Args:
        obj: A mapped instance.

    Returns:
        ``AuditOptions.object_type``, or the class name when it is not set.
    """
    cls = type(obj)
    return options_of(cls).object_type or cls.__name__

field_policy

field_policy(
    model: type[Any], key: str, column: Column[Any]
) -> FieldPolicy | None

Return the audit policy set on a column.

Parameters:

Name	Type	Description	Default
model	type[Any]	The mapped class, named in the error message.	required
key	str	The column's attribute key, named in the error message.	required
column	Column[Any]	The column.	required

Returns:

Type	Description
FieldPolicy | None	The policy, or None when the column has none.

Raises:

Type	Description
FieldPolicyError	info["audit"] is not a known policy.

Source code in audit_trail/diff.py

def field_policy(model: type[Any], key: str, column: Column[Any]) -> FieldPolicy | None:
    """Return the audit policy set on a column.

    Args:
        model: The mapped class, named in the error message.
        key: The column's attribute key, named in the error message.
        column: The column.

    Returns:
        The policy, or ``None`` when the column has none.

    Raises:
        FieldPolicyError: ``info["audit"]`` is not a known policy.
    """
    raw = column.info.get("audit")
    if raw is None:
        return None
    policy = _POLICIES.get(raw) if isinstance(raw, str) else None
    if policy is None:
        raise FieldPolicyError(
            f"{model.__qualname__}.{key}: unknown audit policy {raw!r}; "
            f"expected one of {', '.join(map(repr, _POLICIES))}"
        )
    return policy

entity_changes

entity_changes(
    obj: object,
    kind: ChangeKind,
    *,
    keys: KeyRing | None = None,
    json_encoder: type[JSONEncoder] | None = None,
    global_redact: Collection[str] = (),
) -> Changes

Build the changes of an entity.* entry from an instance.

Meant for after_flush, while session.new / dirty / deleted and attribute history still describe the flush. Reads only what is in memory and never emits SQL.

• created: every audited column as [None, value], empty ones too.
• updated: only columns with a net change. Old and new raw values are compared with column.type.compare_values before any redaction or encoding, so Decimal("1.5") vs Decimal("1.50") is no change. An empty result means there is nothing to record.
• deleted: every audited column as [value, None].

exclude columns are left out. redact stores "***" and hash stores hv{n}:<hex>; both keep null as null. Every other value goes through encode_value.

"<unknown>" is a marker, not a value: it stands where the value is not in memory. That is the old value of a JSON column changed in place (MutableDict) without snapshot_on_load, a deleted column that is not loaded (for example deferred), and a created column filled by a server-side default (server_default, Computed, Identity) that was not fetched back. On PostgreSQL the mapper's default eager_defaults="auto" fetches them with RETURNING on INSERT, so created entries hold the real values; only a mapper with eager_defaults=False leaves them "<unknown>". A created column that was not set and has no server-side default is None, as stored.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required
kind	ChangeKind	Which change set to build.	required
keys	KeyRing | None	Key ring for hash columns.	None
json_encoder	type[JSONEncoder] | None	Host encoder for types encode_value does not handle.	None
global_redact	Collection[str]	Attribute keys or column names redacted when the column has no explicit policy. An extra safety net, not a replacement for per-column policies.	()

Returns:

Type	Description
Changes	{attribute key: [old, new]} with JSON-ready values.

Raises:

Type	Description
FieldPolicyError	A column has an unknown policy, or the model has a hash column and keys is None.
UnserializableValueError	A value cannot be encoded.

Source code in audit_trail/diff.py

def entity_changes(
    obj: object,
    kind: ChangeKind,
    *,
    keys: KeyRing | None = None,
    json_encoder: type[json.JSONEncoder] | None = None,
    global_redact: Collection[str] = (),
) -> Changes:
    """Build the ``changes`` of an ``entity.*`` entry from an instance.

    Meant for ``after_flush``, while ``session.new`` / ``dirty`` / ``deleted``
    and attribute history still describe the flush. Reads only what is in
    memory and never emits SQL.

    - ``created``: every audited column as ``[None, value]``, empty ones too.
    - ``updated``: only columns with a net change. Old and new raw values are
      compared with ``column.type.compare_values`` before any redaction or
      encoding, so ``Decimal("1.5")`` vs ``Decimal("1.50")`` is no change.
      An empty result means there is nothing to record.
    - ``deleted``: every audited column as ``[value, None]``.

    ``exclude`` columns are left out. ``redact`` stores ``"***"`` and
    ``hash`` stores ``hv{n}:<hex>``; both keep null as null. Every other value
    goes through ``encode_value``.

    ``"<unknown>"`` is a marker, not a value: it stands where the value is not
    in memory. That is the old value of a JSON column changed in place
    (``MutableDict``) without ``snapshot_on_load``, a ``deleted`` column that
    is not loaded (for example ``deferred``), and a ``created`` column filled
    by a server-side default (``server_default``, ``Computed``, ``Identity``)
    that was not fetched back. On PostgreSQL the mapper's default
    ``eager_defaults="auto"`` fetches them with ``RETURNING`` on INSERT, so
    ``created`` entries hold the real values; only a mapper with
    ``eager_defaults=False`` leaves them ``"<unknown>"``. A ``created`` column
    that was not set and has no server-side default is ``None``, as stored.

    Args:
        obj: A mapped instance.
        kind: Which change set to build.
        keys: Key ring for ``hash`` columns.
        json_encoder: Host encoder for types ``encode_value`` does not handle.
        global_redact: Attribute keys or column names redacted when the column
            has no explicit policy. An extra safety net, not a replacement for
            per-column policies.

    Returns:
        ``{attribute key: [old, new]}`` with JSON-ready values.

    Raises:
        FieldPolicyError: A column has an unknown policy, or the model has a
            ``hash`` column and ``keys`` is ``None``.
        UnserializableValueError: A value cannot be encoded.
    """
    state = instance_state(obj)
    columns = _audited_columns(state.mapper, frozenset(global_redact))
    if keys is None:
        for spec in columns:
            if spec.policy == "hash":
                raise FieldPolicyError(
                    f"{state.class_.__qualname__}.{spec.key}: the hash policy "
                    "needs a key ring (pseudonymize_key)"
                )

    def render(value: object, spec: _AuditedColumn) -> JSONValue:
        return _render(value, spec.policy, keys, json_encoder)

    changes: Changes = {}
    match kind:
        case "created":
            for spec in columns:
                new = state.dict.get(spec.key, _MISSING)
                if new is _MISSING and spec.column.server_default is None:
                    # Identity and Computed also live in server_default.
                    new = None
                changes[spec.key] = [None, render(new, spec)]
        case "deleted":
            for spec in columns:
                old = state.dict.get(spec.key, _MISSING)
                changes[spec.key] = [render(old, spec), None]
        case "updated":
            snapshot: dict[str, object] = state.info.get(SNAPSHOT_INFO_KEY, {})
            for spec in columns:
                history = state.attrs[spec.key].history
                if not history.added and not history.deleted:
                    continue
                new = history.added[0] if history.added else None
                if history.deleted:
                    old = history.deleted[0]
                else:
                    # Changed in place (MutableDict): the committed value is
                    # gone unless snapshot_on_load kept a copy.
                    old = snapshot.get(spec.key, _MISSING)
                if old is not _MISSING and spec.column.type.compare_values(old, new):
                    continue
                changes[spec.key] = [render(old, spec), render(new, spec)]
        case _:
            assert_never(kind)
    return changes

object_id_of

object_id_of(obj: object) -> str

Return the object_id of an instance.

Uses the identity key when the instance has one. Inside after_flush that is still the key from before the flush, so a changed primary key yields the old id. Objects inserted by the flush have no identity key yet; their primary key is read from the instance.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required

Returns:

Type	Description
str	str(pk) for a single-column key (a UUID in canonical lowercase
str	form), a compact JSON array of strings (["a","1"]) for a
str	composite one.

Raises:

Type	Description
ValueError	The instance has no primary key value yet.

Source code in audit_trail/diff.py

def object_id_of(obj: object) -> str:
    """Return the ``object_id`` of an instance.

    Uses the identity key when the instance has one. Inside ``after_flush``
    that is still the key from before the flush, so a changed primary key
    yields the old id. Objects inserted by the flush have no identity key yet;
    their primary key is read from the instance.

    Args:
        obj: A mapped instance.

    Returns:
        ``str(pk)`` for a single-column key (a UUID in canonical lowercase
        form), a compact JSON array of strings (``["a","1"]``) for a
        composite one.

    Raises:
        ValueError: The instance has no primary key value yet.
    """
    object_id = state_object_id(instance_state(obj))
    if object_id is None:
        raise ValueError(f"{type(obj).__qualname__} instance has no primary key yet")
    return object_id

state_object_id

state_object_id(state: InstanceState[Any]) -> str | None

Return the object_id of an instance state, as :func:object_id_of.

Parameters:

Name	Type	Description	Default
state	InstanceState[Any]	The state of a mapped instance.	required

Returns:

Type	Description
str | None	The id, or None when the instance has no primary key value yet.

Source code in audit_trail/diff.py

def state_object_id(state: InstanceState[Any]) -> str | None:
    """Return the ``object_id`` of an instance state, as :func:`object_id_of`.

    Args:
        state: The state of a mapped instance.

    Returns:
        The id, or ``None`` when the instance has no primary key value yet.
    """
    if state.key is not None:
        return _format_identity(state.key[1])
    mapper = state.mapper
    values = tuple(
        state.dict.get(mapper.get_property_by_column(column).key)
        for column in mapper.primary_key
    )
    if any(value is None for value in values):
        return None
    return _format_identity(values)

object_id_for

object_id_for(model: type[Any], pk: object) -> str

Return the object_id for a primary key of model.

Parameters:

Name	Type	Description	Default
model	type[Any]	A mapped class.	required
pk	object	The key value, or a tuple of values in primary key column order for a composite key.	required

Returns:

Type	Description
str	The same string object_id_of returns for that instance.

Raises:

Type	Description
ValueError	The number of values does not match the primary key, or a value is None.

Source code in audit_trail/diff.py

def object_id_for(model: type[Any], pk: object) -> str:
    """Return the ``object_id`` for a primary key of ``model``.

    Args:
        model: A mapped class.
        pk: The key value, or a tuple of values in primary key column order
            for a composite key.

    Returns:
        The same string ``object_id_of`` returns for that instance.

    Raises:
        ValueError: The number of values does not match the primary key, or
            a value is ``None``.
    """
    mapper: Mapper[Any] = inspect(model)
    values = pk if isinstance(pk, tuple) else (pk,)
    if len(values) != len(mapper.primary_key):
        raise ValueError(
            f"{model.__qualname__} has a {len(mapper.primary_key)}-column primary "
            f"key, got {len(values)} value(s)"
        )
    if any(value is None for value in values):
        raise ValueError(f"primary key of {model.__qualname__} contains None")
    return _format_identity(values)

resolve_label

resolve_label(
    obj: object, options: AuditOptions
) -> str | None

Evaluate options.label for object_label without emitting SQL.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required
options	AuditOptions	The model's options.	required

Returns:

Type	Description
str | None	The label as a string, or None when there is no label option,
str | None	it returns None, or it read an attribute that is not loaded (then
str | None	logged once per model, attribute and option on the
str | None	audit_trail.diff logger).

Raises:

Type	Description
AuditOptionError	The option read a functools.cached_property.

Source code in audit_trail/diff.py

def resolve_label(obj: object, options: AuditOptions) -> str | None:
    """Evaluate ``options.label`` for ``object_label`` without emitting SQL.

    Args:
        obj: A mapped instance.
        options: The model's options.

    Returns:
        The label as a string, or ``None`` when there is no ``label`` option,
        it returns ``None``, or it read an attribute that is not loaded (then
        logged once per model, attribute and option on the
        ``audit_trail.diff`` logger).

    Raises:
        AuditOptionError: The option read a ``functools.cached_property``.
    """
    if options.label is None:
        return None
    ok, value = _call_option(obj, "label", options.label)
    return str(value) if ok and value is not None else None

resolve_scope

resolve_scope(
    obj: object, options: AuditOptions
) -> str | UseContext | None

Evaluate options.scope for scope_id without emitting SQL.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required
options	AuditOptions	The model's options.	required

Returns:

Type	Description
str | UseContext | None	USE_CONTEXT when there is no scope option or it read an
str | UseContext | None	attribute that is not loaded (then logged once per model, attribute
str | UseContext | None	and option on the audit_trail.diff logger). Otherwise None
str | UseContext | None	when the option returned None (an explicit "no scope"), else the
str | UseContext | None	value as a string.

Raises:

Type	Description
AuditOptionError	The option read a functools.cached_property.

Source code in audit_trail/diff.py

def resolve_scope(obj: object, options: AuditOptions) -> str | UseContext | None:
    """Evaluate ``options.scope`` for ``scope_id`` without emitting SQL.

    Args:
        obj: A mapped instance.
        options: The model's options.

    Returns:
        ``USE_CONTEXT`` when there is no ``scope`` option or it read an
        attribute that is not loaded (then logged once per model, attribute
        and option on the ``audit_trail.diff`` logger). Otherwise ``None``
        when the option returned ``None`` (an explicit "no scope"), else the
        value as a string.

    Raises:
        AuditOptionError: The option read a ``functools.cached_property``.
    """
    if options.scope is None:
        return USE_CONTEXT
    ok, value = _call_option(obj, "scope", options.scope)
    if not ok:
        return USE_CONTEXT
    return None if value is None else str(value)

resolve_target

resolve_target(
    obj: object, options: AuditOptions
) -> ResolvedTarget | None

Evaluate options.target for target_type/target_id.

Never emits SQL. The id is formatted like object_id: a tuple becomes a composite id, anything else str().

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance.	required
options	AuditOptions	The model's options.	required

Returns:

Type	Description
ResolvedTarget | None	The stored target, or None when there is no target
ResolvedTarget | None	option, it returns None or an id of None, or it read an
ResolvedTarget | None	attribute that is not loaded (then logged once per model, attribute
ResolvedTarget | None	and option on the audit_trail.diff logger).

Raises:

Type	Description
AuditOptionError	The option read a functools.cached_property.

Source code in audit_trail/diff.py

def resolve_target(obj: object, options: AuditOptions) -> ResolvedTarget | None:
    """Evaluate ``options.target`` for ``target_type``/``target_id``.

    Never emits SQL. The id is formatted like ``object_id``: a tuple becomes
    a composite id, anything else ``str()``.

    Args:
        obj: A mapped instance.
        options: The model's options.

    Returns:
        The stored target, or ``None`` when there is no ``target``
        option, it returns ``None`` or an id of ``None``, or it read an
        attribute that is not loaded (then logged once per model, attribute
        and option on the ``audit_trail.diff`` logger).

    Raises:
        AuditOptionError: The option read a ``functools.cached_property``.
    """
    if options.target is None:
        return None
    ok, value = _call_option(obj, "target", options.target)
    if not ok or value is None:
        return None
    return format_target(Target(*value))

format_target

format_target(target: Target) -> ResolvedTarget | None

Format a target as stored in target_type and target_id.

Parameters:

Name	Type	Description	Default
target	Target	The type name and the id; a tuple id is a composite key.	required

Returns:

Type	Description
ResolvedTarget | None	The stored target, or None when the id is None.

Source code in audit_trail/diff.py

def format_target(target: Target) -> ResolvedTarget | None:
    """Format a target as stored in ``target_type`` and ``target_id``.

    Args:
        target: The type name and the id; a tuple id is a composite key.

    Returns:
        The stored target, or ``None`` when the id is ``None``.
    """
    if target.id is None:
        return None
    ids = target.id if isinstance(target.id, tuple) else (target.id,)
    return ResolvedTarget(str(target.type), _format_identity(ids))