audit_trail.relations

Relationship deltas captured from attribute events.

Collection changes are recorded from the ORM's append / remove attribute events, not from attr.history: history is reset by every flush, so a removal written by an autoflush between two collection operations is no longer visible when the unit of work ends. Events see every operation once, and their cost is proportional to the delta, not to the collection size.

Loading a collection from the database (lazy, selectinload, joinedload, refresh) populates it without firing these events, so a load never looks like an edit.

Known limitations: changes that bypass the collection API (writing a foreign key column directly, raw inserts into an association table) are invisible, and a backref that moves an object away from a parent that is not loaded in the session fires no remove on that parent.

RelationshipChange

Bases: TypedDict

Net change of one tracked collection since the last flush.

Attributes:

Name	Type	Description
added	list[str]	Object ids of members added to the collection.
removed	list[str]	Object ids of members removed from it.

track_relationships

track_relationships(
    *attributes: InstrumentedAttribute[Any],
) -> None

Record net added / removed changes of collection relationships.

Registers append and remove listeners on each attribute (also covering assignment, clear(), slicing and del, which emit the same events) and an expire listener on its class. Listeners propagate to subclasses, so registering an attribute already tracked on the same class or a base class is a no-op.

Parameters:

Name	Type	Description	Default
*attributes	InstrumentedAttribute[Any]	Class-bound collection relationships, e.g. Post.tags.	()

Raises:

Type	Description
ValueError	If an attribute is not a collection relationship(), or if the same relationship is already tracked on a subclass.

Source code in audit_trail/relations.py

def track_relationships(*attributes: InstrumentedAttribute[Any]) -> None:
    """Record net ``added`` / ``removed`` changes of collection relationships.

    Registers ``append`` and ``remove`` listeners on each attribute (also
    covering assignment, ``clear()``, slicing and ``del``, which emit the same
    events) and an ``expire`` listener on its class. Listeners propagate to
    subclasses, so registering an attribute already tracked on the same class
    or a base class is a no-op.

    Args:
        *attributes: Class-bound collection relationships, e.g. ``Post.tags``.

    Raises:
        ValueError: If an attribute is not a collection ``relationship()``,
            or if the same relationship is already tracked on a subclass.
    """
    for attr in attributes:
        prop = attr.property
        if not isinstance(prop, RelationshipProperty) or not prop.uselist:
            raise ValueError(f"{attr} is not a collection relationship")
        cls, key = cast("type[Any]", attr.class_), attr.key
        # Listeners propagate to subclasses, so a key tracked on any class in
        # the MRO already covers this one; registering again would count
        # every change twice.
        if any(_TrackedAttribute(base, key) in _tracked for base in cls.__mro__):
            continue
        for tracked in _tracked:
            if tracked.key == key and issubclass(tracked.cls, cls):
                raise ValueError(
                    f"cannot track {cls.__name__}.{key}: subclass attribute "
                    f"{tracked.cls.__name__}.{key} is already tracked and its "
                    "changes would be counted twice"
                )
        _listen(attr, key)
        if not event.contains(cls, "expire", _on_expire):
            event.listen(cls, "expire", _on_expire, raw=True, propagate=True)
        _tracked.add(_TrackedAttribute(cls, key))

pop_relationship_changes

pop_relationship_changes(
    obj: object,
) -> dict[str, RelationshipChange]

Return and clear the net collection changes of obj.

Meant to be called from after_flush or later, when every added item has a primary key. Items that still have none were not written by the flush (SQLAlchemy skips objects not cascaded into the session), so they are left out, matching the database.

Parameters:

Name	Type	Description	Default
obj	object	A mapped instance whose relationships are tracked.	required

Returns:

Type	Description
dict[str, RelationshipChange]	{attribute: {"added": [object_id, ...], "removed": [...]}} for
dict[str, RelationshipChange]	each tracked attribute with a non-empty net change; empty when nothing
dict[str, RelationshipChange]	changed.

Source code in audit_trail/relations.py

def pop_relationship_changes(obj: object) -> dict[str, RelationshipChange]:
    """Return and clear the net collection changes of ``obj``.

    Meant to be called from ``after_flush`` or later, when every added item
    has a primary key. Items that still have none were not written by the flush
    (SQLAlchemy skips objects not cascaded into the session), so they are
    left out, matching the database.

    Args:
        obj: A mapped instance whose relationships are tracked.

    Returns:
        ``{attribute: {"added": [object_id, ...], "removed": [...]}}`` for
        each tracked attribute with a non-empty net change; empty when nothing
        changed.
    """
    state = cast("InstanceState[Any]", inspect(obj))
    store: dict[str, _Delta] | None = state.info.pop(_INFO_KEY, None)
    if not store:
        return {}
    changes: dict[str, RelationshipChange] = {}
    for key, delta in store.items():
        added = _object_ids(delta.added)
        removed = _object_ids(delta.removed)
        if added or removed:
            changes[key] = {"added": added, "removed": removed}
    return changes

discard_relationship_changes

discard_relationship_changes(session: Session) -> None

Drop the pending deltas of every object attached to session.

A belt-and-braces call for rollback listeners: the expire listener already drops deltas whenever the ORM discards unflushed collection state. Pending objects expunged by a rollback keep their deltas, just as they keep their in-memory collections, so re-adding them to a session reports what that session's flush writes.

Parameters:

Name	Type	Description	Default
session	Session	The session being rolled back.	required

Source code in audit_trail/relations.py

def discard_relationship_changes(session: Session) -> None:
    """Drop the pending deltas of every object attached to ``session``.

    A belt-and-braces call for rollback listeners: the ``expire`` listener
    already drops deltas whenever the ORM discards unflushed collection
    state. Pending objects expunged by a rollback keep their deltas, just as
    they keep their in-memory collections, so re-adding them to a session
    reports what that session's flush writes.

    Args:
        session: The session being rolled back.
    """
    for state in session.identity_map.all_states():
        state.info.pop(_INFO_KEY, None)
    for obj in session.new:
        inspect(obj).info.pop(_INFO_KEY, None)