audit_trail.config

Library configuration: AuditTrail and per-model AuditOptions.

DURABLE_POOL_SIZE module-attribute

DURABLE_POOL_SIZE = 5

Connections in the durable engine the library builds.

DURABLE_POOL_TIMEOUT module-attribute

DURABLE_POOL_TIMEOUT = 5.0

Seconds a durable write waits for a connection of that engine.

Target

Bases: NamedTuple

A parent object referred to by type name and primary key.

Returned by the target option and accepted by AuditTrail.log. A plain (type, id) tuple works too.

Attributes:

Name	Type	Description
type	str	Name stored in target_type.
id	object	The primary key; a tuple for a composite key. None means no target.

AuditOptions dataclass

AuditOptions(
    severity: IntEnum | None = None,
    verb_severity: dict[str, IntEnum] = dict(),
    track_relationships: Collection[str] = frozenset(),
    snapshot_on_load: Collection[str] = frozenset(),
    object_type: str | None = None,
    label: Callable[[Any], str | None] | None = None,
    scope: Callable[[Any], object] | None = None,
    target: Callable[
        [Any], Target | tuple[str, object] | None
    ]
    | None = None,
)

Audit settings for one model, set as __audit__ on the class.

label, scope and target run while the session flushes and get a read-only view of the instance instead of the instance itself. They may read only attributes already loaded on it: reading an expired or unloaded attribute emits no SQL, logs a warning on audit_trail.diff (once per model, attribute and option) and the option falls back (label and target to None, scope to the context). A column that was never set on a new instance reads as None, as stored, unless it has a server-side default. Reading a functools.cached_property raises AuditOptionError. Any other exception they raise propagates and aborts the flush.

Attributes:

Name	Type	Description
severity	IntEnum | None	Severity of the model's entity.* entries. None uses the configured default.
verb_severity	dict[str, IntEnum]	Per-verb severity overrides, keyed by verb (for example {"entity.deleted": MySeverity.HIGH}).
track_relationships	Collection[str]	Relationship names whose membership changes are recorded as {"added": [...], "removed": [...]}.
snapshot_on_load	Collection[str]	JSON columns deep-copied on load and refresh, so an in-place mutation still has an old value.
object_type	str | None	Name stored in object_type. Defaults to the class name.
label	Callable[[Any], str | None] | None	Returns the object's label, stored in object_label.
scope	Callable[[Any], object] | None	Returns the scope id, stored in scope_id. Returning None means "no scope"; the scope comes from the context only when this option is not set or it read an attribute that is not loaded.
target	Callable[[Any], Target | tuple[str, object] | None] | None	Returns the parent object as a Target (or a plain (type, id) tuple), or None. An id of None also means no target; a tuple id is formatted as a composite object_id.

AuditTrail

AuditTrail(
    engine: Engine | AsyncEngine,
    *,
    schema: str = "audit",
    severities: type[IntEnum] | None = None,
    default_severity: IntEnum | None = None,
    system_severity: IntEnum | None = None,
    events: Iterable[type[AuditEvent]] | None = None,
    durable_engine: Engine | AsyncEngine | None = None,
    on_error: OnError = "log",
    pseudonymize_key: bytes
    | dict[int, bytes]
    | None = None,
    session_provider: Callable[
        [], Session | AsyncSession | None
    ]
    | None = None,
    context_provider: Callable[[], Any] | None = None,
    json_encoder: type[JSONEncoder] | None = None,
    indexes: Collection[str] | None = None,
    global_redact: Collection[str] = (),
    warn_on_bulk: bool = True,
    auto_create_partitions: bool = False,
    allow_scrub: bool = False,
    default_query_window: timedelta | None = None,
)

Entry point of the library: configuration plus the public operations.

Parameters:

Name	Type	Description	Default
engine	Engine | AsyncEngine	Sync or async engine used for DDL and maintenance.	required
schema	str	Database schema holding the audit tables.	'audit'
severities	type[IntEnum] | None	The application's severity IntEnum. None uses the built-in Severity.	None
default_severity	IntEnum | None	Severity of entity.* entries whose model sets none. None uses the lowest member of severities.	None
system_severity	IntEnum | None	Severity of the library's own audit.* events. None uses the highest member of severities.	None
events	Iterable[type[AuditEvent]] | None	Event classes to register. None registers every AuditEvent subclass with members defined when the AuditTrail is created, so import them first or list them here.	None
durable_engine	Engine | AsyncEngine | None	Engine for durable writes: sync for log, async for alog. Must not be configured for AUTOCOMMIT. None builds one from engine's URL (only the URL: pass your own engine for connect_args and similar settings), sync or async like engine, with its own small pool (DURABLE_POOL_SIZE connections, no overflow, DURABLE_POOL_TIMEOUT, pre-ping). It connects on the first durable write; dispose or adispose closes it.	None
on_error	OnError	"log" keeps the business transaction alive when an audit write fails; "raise" propagates. fail_closed events always raise.	'log'
pseudonymize_key	bytes | dict[int, bytes] | None	HMAC key as bytes (version 1) or {version: key}.	None
session_provider	Callable[[], Session | AsyncSession | None] | None	Returns the current session, or None when there is none, for hosts that keep it in a context variable. log and alog call it when no session is passed.	None
context_provider	Callable[[], Any] | None	Returns the current audit context.	None
json_encoder	type[JSONEncoder] | None	Extra encoder for types the built-in one does not handle.	None
indexes	Collection[str] | None	Names of the indexes to create. None uses the defaults.	None
global_redact	Collection[str]	Attribute keys or column names redacted in entity.* changes when the column has no explicit policy. A safety net, not a replacement for per-column policies.	()
warn_on_bulk	bool	Log a warning when a bulk update()/delete() hits an audited table.	True
auto_create_partitions	bool	Let durable writes create a missing partition and retry once. Needs DDL privileges.	False
allow_scrub	bool	Enable scrub/scrub_actor. They run on engine, whose role then needs UPDATE on the audit tables.	False
default_query_window	timedelta | None	How far back query.list_groups reads when it is not given since, for example timedelta(days=30); since=ALL_HISTORY still lists the whole log. None reads the whole log, which gets slower as it grows.	None

Attributes:

Name	Type	Description
severities	type[IntEnum]	The severity enum in use.
registry		The registered events.
tables		The audit tables built from schema and indexes.
keys		Key ring built from pseudonymize_key, or None.
maintenance		Partition management on engine, for example audit.maintenance.ensure_partitions().
query		Read queries, for example audit.query.list_groups(session).
durable_engine	Engine | AsyncEngine	The engine durable writes use.

Raises:

Type	Description
EventRegistryError	An event or a severity setting is invalid.
ValueError	A table name or an index key is invalid, pseudonymize_key is too short, or default_query_window is not positive.
TypeError	pseudonymize_key or default_query_window has the wrong type.

Source code in audit_trail/config.py

def __init__(
    self,
    engine: Engine | AsyncEngine,
    *,
    schema: str = "audit",
    severities: type[IntEnum] | None = None,
    default_severity: IntEnum | None = None,
    system_severity: IntEnum | None = None,
    events: Iterable[type[AuditEvent]] | None = None,
    durable_engine: Engine | AsyncEngine | None = None,
    on_error: OnError = "log",
    pseudonymize_key: bytes | dict[int, bytes] | None = None,
    session_provider: Callable[[], Session | AsyncSession | None] | None = None,
    context_provider: Callable[[], Any] | None = None,
    json_encoder: type[json.JSONEncoder] | None = None,
    indexes: Collection[str] | None = None,
    global_redact: Collection[str] = (),
    warn_on_bulk: bool = True,
    auto_create_partitions: bool = False,
    allow_scrub: bool = False,
    default_query_window: timedelta | None = None,
) -> None:
    self.engine = engine
    self.schema = schema
    self.severities: type[IntEnum] = Severity if severities is None else severities
    self.registry = EventRegistry(
        self.severities,
        events=events,
        default_severity=default_severity,
        system_severity=system_severity,
    )
    self._owns_durable_engine = durable_engine is None
    self.durable_engine: Engine | AsyncEngine = (
        _default_durable_engine(engine)
        if durable_engine is None
        else durable_engine
    )
    self.on_error: OnError = on_error
    self.pseudonymize_key = pseudonymize_key
    self.session_provider = session_provider
    self.context_provider = context_provider
    self.json_encoder = json_encoder
    self.indexes = indexes
    self.global_redact = frozenset(global_redact)
    self.warn_on_bulk = warn_on_bulk
    self.auto_create_partitions = auto_create_partitions
    self.allow_scrub = allow_scrub
    self.keys = None if pseudonymize_key is None else KeyRing(pseudonymize_key)
    self.tables = build_tables(schema=schema, indexes=indexes)
    self.maintenance = PartitionManager(engine, self.tables, self.severities)
    from audit_trail.query import AuditQuery

    self.query = AuditQuery(
        self.tables, self.severities, default_window=default_query_window
    )

install

install(
    session_factory: type[Session | AsyncSession]
    | sessionmaker[Any]
    | async_sessionmaker[Any],
) -> None

Audit the sessions of a factory.

Registers the listeners on the factory's session class only (for a sessionmaker, the class it builds), so other sessions, a second sessionmaker and tools such as Alembic are not audited. Set session.info["audit_enabled"] = False to switch capture off for one session of that class.

An AsyncSession runs a sync Session in a greenlet, and the listeners are registered on that class: give the async factory your own Session subclass, as in async_sessionmaker(engine, sync_session_class=AppSession) or as sync_session_class of an AsyncSession subclass.

Parameters:

Name	Type	Description	Default
session_factory	type[Session | AsyncSession] | sessionmaker[Any] | async_sessionmaker[Any]	A sessionmaker or Session subclass, or an async_sessionmaker or AsyncSession subclass.	required

Raises:

Type	Description
TypeError	session_factory is none of these, or its sync_session_class is not a Session subclass.
ValueError	The session class is the base Session (for an async factory: no sync_session_class was given), or an AuditTrail is already installed on it, a base class or a subclass of it.

Source code in audit_trail/config.py

def install(
    self,
    session_factory: type[Session | AsyncSession]
    | sessionmaker[Any]
    | async_sessionmaker[Any],
) -> None:
    """Audit the sessions of a factory.

    Registers the listeners on the factory's session class only (for a
    ``sessionmaker``, the class it builds), so other sessions, a second
    ``sessionmaker`` and tools such as Alembic are not audited. Set
    ``session.info["audit_enabled"] = False`` to switch capture off for
    one session of that class.

    An ``AsyncSession`` runs a sync ``Session`` in a greenlet, and the
    listeners are registered on that class: give the async factory your
    own ``Session`` subclass, as in
    ``async_sessionmaker(engine, sync_session_class=AppSession)`` or as
    ``sync_session_class`` of an ``AsyncSession`` subclass.

    Args:
        session_factory: A ``sessionmaker`` or ``Session`` subclass, or
            an ``async_sessionmaker`` or ``AsyncSession`` subclass.

    Raises:
        TypeError: ``session_factory`` is none of these, or its
            ``sync_session_class`` is not a ``Session`` subclass.
        ValueError: The session class is the base ``Session`` (for an
            async factory: no ``sync_session_class`` was given), or an
            ``AuditTrail`` is already installed on it, a base class or a
            subclass of it.
    """
    from audit_trail.listener import install

    install(self, session_factory)

pseudonymize

pseudonymize(value: object, *, purpose: str) -> str

Replace a value with a keyed pseudonym, e.g. audit.<purpose>.v1:<hex>.

Parameters:

Name	Type	Description	Default
value	object	The value. Must not be None.	required
purpose	str	Snake_case name separating unrelated uses.	required

Returns:

Type	Description
str	The pseudonym token.

Raises:

Type	Description
ValueError	No pseudonymize_key is configured, or purpose is not snake_case.
TypeError	value is None.
UnserializableValueError	value cannot be encoded.

Source code in audit_trail/config.py

def pseudonymize(self, value: object, *, purpose: str) -> str:
    """Replace a value with a keyed pseudonym, e.g. ``audit.<purpose>.v1:<hex>``.

    Args:
        value: The value. Must not be ``None``.
        purpose: Snake_case name separating unrelated uses.

    Returns:
        The pseudonym token.

    Raises:
        ValueError: No ``pseudonymize_key`` is configured, or ``purpose``
            is not snake_case.
        TypeError: ``value`` is ``None``.
        UnserializableValueError: ``value`` cannot be encoded.
    """
    if self.keys is None:
        raise ValueError("pseudonymize needs pseudonymize_key")
    return _pseudonymize(
        value, purpose=purpose, keys=self.keys, json_encoder=self.json_encoder
    )

dispose

dispose() -> None

Close the pooled connections of a durable engine the library built.

A durable_engine passed in is the host's to dispose; this leaves it alone.

Raises:

Type	Description
TypeError	The library's durable engine is async; use adispose.

Source code in audit_trail/config.py

def dispose(self) -> None:
    """Close the pooled connections of a durable engine the library built.

    A ``durable_engine`` passed in is the host's to dispose; this leaves it
    alone.

    Raises:
        TypeError: The library's durable engine is async; use ``adispose``.
    """
    if not self._owns_durable_engine:
        return
    if not isinstance(self.durable_engine, Engine):
        raise TypeError("the durable engine is async; use adispose")
    self.durable_engine.dispose()

adispose async

adispose() -> None

Async dispose, for a durable engine the library built async.

Raises:

Type	Description
TypeError	The library's durable engine is sync; use dispose.

Source code in audit_trail/config.py

async def adispose(self) -> None:
    """Async ``dispose``, for a durable engine the library built async.

    Raises:
        TypeError: The library's durable engine is sync; use ``dispose``.
    """
    if not self._owns_durable_engine:
        return
    if isinstance(self.durable_engine, Engine):
        raise TypeError("the durable engine is sync; use dispose")
    await self.durable_engine.dispose()

log

log(
    session: Session | None,
    event: AuditEvent,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

log(
    event: AuditEvent,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

log(
    session_or_event: Session | AuditEvent | None,
    event: AuditEvent | None = None,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

Record an explicit event.

Called as log(session, event, ...) or, with a session_provider configured, as log(event, ...): without a session (or with None), the entry goes to the session session_provider() returns.

A non-durable entry is inserted immediately on the session's connection, in the same database transaction as the session's changes: it is committed or rolled back with them, and shares their audit_transaction row. Write failures follow on_error. With obj, the connection is the one obj is flushed on; without, the one for the audit tables: their bind in Session(binds=...), else the session's default bind.

A durable entry (the event's durable or fail_closed, or durable=True) is written on a connection of durable_engine in a transaction of its own, with its own audit_transaction row built from the current context, and committed before this returns, so it survives a rollback of the session. A failed write is logged with on_error="log"; with on_error="raise", and always for a fail_closed event, it raises AuditWriteError. With auto_create_partitions, a missing partition is created and the write retried once. Creating a partition waits for every open transaction that has written audit rows, including the session's own: if it has, the retry fails after the ensure_partitions lock timeout (about 5 to 10 seconds) and a fail_closed event raises AuditWriteError. Run ensure_partitions ahead of time rather than relying on the retry.

Either kind is written even when session.info["audit_enabled"] is False, which only switches off the automatic entity.* entries.

Parameters:

Name	Type	Description	Default
session_or_event	Session | AuditEvent | None	A session of a class this AuditTrail is installed on, followed by the event; or the event alone, or after None, to use session_provider.	required
event	AuditEvent | None	A registered host event.	None
obj	object | None	The object the event is about; sets object_type, object_id, object_label and, through its options, scope_id and the default target. It must have a primary key: flush a new object first.	None
target	Target | object | None	The parent object, as an instance or a Target (a plain (type, id) tuple works too). A tuple id is formatted as a composite object_id; an id of None means no target. Defaults to the target option of obj.	None
payload	Mapping[str, object] | BaseModel | None	The payload, validated against the event's schema. Pseudonymized fields of a schema are pseudonymized with their field name as the purpose.	None
actor	Actor | None	The actor of this entry. It sets the entry's actor_id and the actor fields of its data.context; the audit_transaction row keeps the actor of the context.	None
durable	bool | None	Overrides the event's durable setting; False cannot weaken a fail_closed event.	None

Raises:

Type	Description
RuntimeError	No session is passed and there is no session_provider, or it returns None.
TypeError	The arguments are not (session, event) or (event), the session is not a Session (use alog for an AsyncSession), its class is not installed by this AuditTrail, or the write is durable and durable_engine is async.
ValueError	The event uses a reserved prefix, durable=False is given for a fail_closed event, obj or an instance target has no primary key, or a Pseudonymized field cannot be pseudonymized.
UnknownEventError	The event is not registered.
PayloadError	The payload does not match the event's schema, or a Pseudonymized field already holds a pseudonym token.
AuditWriteError	A durable write failed and the policy says to raise.
UnboundExecutionError	A non-durable entry without obj, and the session has neither a bind for the audit tables nor a default bind.

Source code in audit_trail/config.py

def log(
    self,
    session_or_event: Session | AuditEvent | None,
    event: AuditEvent | None = None,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None:
    """Record an explicit event.

    Called as ``log(session, event, ...)`` or, with a ``session_provider``
    configured, as ``log(event, ...)``: without a session (or with
    ``None``), the entry goes to the session ``session_provider()``
    returns.

    A non-durable entry is inserted immediately on the session's
    connection, in the same database transaction as the session's changes:
    it is committed or rolled back with them, and shares their
    ``audit_transaction`` row. Write failures follow ``on_error``. With
    ``obj``, the connection is the one ``obj`` is flushed on; without, the
    one for the audit tables: their bind in ``Session(binds=...)``, else
    the session's default bind.

    A durable entry (the event's ``durable`` or ``fail_closed``, or
    ``durable=True``) is written on a connection of ``durable_engine`` in
    a transaction of its own, with its own ``audit_transaction`` row built
    from the current context, and committed before this returns, so it
    survives a rollback of the session. A failed write is logged with
    ``on_error="log"``; with ``on_error="raise"``, and always for a
    ``fail_closed`` event, it raises ``AuditWriteError``. With
    ``auto_create_partitions``, a missing partition is created and the
    write retried once. Creating a partition waits for every open
    transaction that has written audit rows, including the session's own:
    if it has, the retry fails after the ``ensure_partitions`` lock
    timeout (about 5 to 10 seconds) and a ``fail_closed`` event raises
    ``AuditWriteError``. Run ``ensure_partitions`` ahead of time rather
    than relying on the retry.

    Either kind is written even when ``session.info["audit_enabled"]`` is
    ``False``, which only switches off the automatic ``entity.*`` entries.

    Args:
        session_or_event: A session of a class this ``AuditTrail`` is
            installed on, followed by the event; or the event alone, or
            after ``None``, to use ``session_provider``.
        event: A registered host event.
        obj: The object the event is about; sets ``object_type``,
            ``object_id``, ``object_label`` and, through its options,
            ``scope_id`` and the default target. It must have a primary
            key: flush a new object first.
        target: The parent object, as an instance or a ``Target`` (a
            plain ``(type, id)`` tuple works too). A tuple id is formatted
            as a composite ``object_id``; an id of ``None`` means no
            target. Defaults to the ``target`` option of ``obj``.
        payload: The payload, validated against the event's schema.
            ``Pseudonymized`` fields of a schema are pseudonymized with
            their field name as the purpose.
        actor: The actor of this entry. It sets the entry's ``actor_id``
            and the actor fields of its ``data.context``; the
            ``audit_transaction`` row keeps the actor of the context.
        durable: Overrides the event's ``durable`` setting; ``False``
            cannot weaken a ``fail_closed`` event.

    Raises:
        RuntimeError: No session is passed and there is no
            ``session_provider``, or it returns ``None``.
        TypeError: The arguments are not ``(session, event)`` or
            ``(event)``, the session is not a ``Session`` (use ``alog``
            for an ``AsyncSession``), its class is not installed by this
            ``AuditTrail``, or the write is durable and ``durable_engine``
            is async.
        ValueError: The event uses a reserved prefix, ``durable=False``
            is given for a ``fail_closed`` event, ``obj`` or an instance
            ``target`` has no primary key, or a ``Pseudonymized`` field
            cannot be pseudonymized.
        UnknownEventError: The event is not registered.
        PayloadError: The payload does not match the event's schema, or a
            ``Pseudonymized`` field already holds a pseudonym token.
        AuditWriteError: A durable write failed and the policy says to
            raise.
        UnboundExecutionError: A non-durable entry without ``obj``, and
            the session has neither a bind for the audit tables nor a
            default bind.
    """
    from sqlalchemy.orm import Session

    from audit_trail.listener import PendingEntry, write_entries
    from audit_trail.writer import (
        DURABLE_WRITE_ERRORS,
        handle_durable_failure,
        write_durable,
    )

    call = _split_call("log", session_or_event, event)
    session = self._session_for("log", call.session)
    if not isinstance(session, Session):
        raise TypeError(
            f"log() needs a Session, not {type(session).__qualname__}; "
            "use alog() for an AsyncSession"
        )
    entry = self._prepare(session, call.event, obj, target, payload, actor, durable)
    if not entry.flags.durable:
        write_entries(session, self, [PendingEntry(entry.entry, obj)], entry.ctx)
        return
    engine = self.durable_engine
    if not isinstance(engine, Engine):
        raise TypeError(
            "log() writes durable entries with a sync durable_engine; "
            "this one is async: use alog()"
        )
    values = self._transaction_values(entry.ctx)
    try:
        with engine.connect() as conn:
            write_durable(
                conn,
                self.tables,
                self._severity_values(),
                values,
                [entry.entry],
                auto_create_partitions=self.auto_create_partitions,
            )
    except DURABLE_WRITE_ERRORS as exc:
        handle_durable_failure(exc, self.on_error, entry.flags.fail_closed)

alog async

alog(
    session: AsyncSession | None,
    event: AuditEvent,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

alog(
    event: AuditEvent,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

alog(
    session_or_event: AsyncSession | AuditEvent | None,
    event: AuditEvent | None = None,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None

Async log, for an AsyncSession of an installed factory.

Called as alog(session, event, ...) or, with a session_provider configured, as alog(event, ...); see log.

A non-durable entry is written in the session's transaction, a durable one on the async durable_engine; see log.

Parameters:

Name	Type	Description	Default
session_or_event	AsyncSession | AuditEvent | None	An AsyncSession whose sync_session_class this AuditTrail is installed on, followed by the event; or the event alone, or after None, to use session_provider.	required
event	AuditEvent | None	A registered host event.	None
obj	object | None	The object the event is about; see log.	None
target	Target | object | None	The parent object; see log.	None
payload	Mapping[str, object] | BaseModel | None	The payload; see log.	None
actor	Actor | None	The actor of this entry; see log.	None
durable	bool | None	Overrides the event's durable setting.	None

Raises:

Type	Description
RuntimeError	No session is passed and there is no session_provider, or it returns None.
TypeError	The arguments are not (session, event) or (event), the session is not an AsyncSession (use log for a Session), its class is not installed by this AuditTrail, or the write is durable and durable_engine is sync.
ValueError	As for log.
UnknownEventError	The event is not registered.
PayloadError	The payload does not match the event's schema.
AuditWriteError	A durable write failed and the policy says to raise.
UnboundExecutionError	As for log.

Source code in audit_trail/config.py

async def alog(
    self,
    session_or_event: AsyncSession | AuditEvent | None,
    event: AuditEvent | None = None,
    /,
    *,
    obj: object | None = None,
    target: Target | object | None = None,
    payload: Mapping[str, object] | BaseModel | None = None,
    actor: Actor | None = None,
    durable: bool | None = None,
) -> None:
    """Async ``log``, for an ``AsyncSession`` of an installed factory.

    Called as ``alog(session, event, ...)`` or, with a
    ``session_provider`` configured, as ``alog(event, ...)``; see ``log``.

    A non-durable entry is written in the session's transaction, a
    durable one on the async ``durable_engine``; see ``log``.

    Args:
        session_or_event: An ``AsyncSession`` whose ``sync_session_class``
            this ``AuditTrail`` is installed on, followed by the event; or
            the event alone, or after ``None``, to use
            ``session_provider``.
        event: A registered host event.
        obj: The object the event is about; see ``log``.
        target: The parent object; see ``log``.
        payload: The payload; see ``log``.
        actor: The actor of this entry; see ``log``.
        durable: Overrides the event's ``durable`` setting.

    Raises:
        RuntimeError: No session is passed and there is no
            ``session_provider``, or it returns ``None``.
        TypeError: The arguments are not ``(session, event)`` or
            ``(event)``, the session is not an ``AsyncSession`` (use
            ``log`` for a ``Session``), its class is not installed by this
            ``AuditTrail``, or the write is durable and ``durable_engine``
            is sync.
        ValueError: As for ``log``.
        UnknownEventError: The event is not registered.
        PayloadError: The payload does not match the event's schema.
        AuditWriteError: A durable write failed and the policy says to
            raise.
        UnboundExecutionError: As for ``log``.
    """
    from sqlalchemy.ext.asyncio import AsyncSession

    from audit_trail.listener import PendingEntry, write_entries
    from audit_trail.writer import (
        async_durable_write_errors,
        handle_durable_failure,
        write_durable,
    )

    call = _split_call("alog", session_or_event, event)
    session = self._session_for("alog", call.session)
    if not isinstance(session, AsyncSession):
        raise TypeError(
            f"alog() needs an AsyncSession, not {type(session).__qualname__}; "
            "use log() for a Session"
        )
    entry = self._prepare(
        session.sync_session, call.event, obj, target, payload, actor, durable
    )
    if not entry.flags.durable:
        await session.run_sync(
            lambda sync_session: write_entries(
                sync_session, self, [PendingEntry(entry.entry, obj)], entry.ctx
            )
        )
        return
    engine = self.durable_engine
    if isinstance(engine, Engine):
        raise TypeError(
            "alog() writes durable entries with an async durable_engine; "
            "this one is sync: use log()"
        )
    values = self._transaction_values(entry.ctx)
    try:
        async with engine.connect() as conn:
            await conn.run_sync(
                write_durable,
                self.tables,
                self._severity_values(),
                values,
                [entry.entry],
                auto_create_partitions=self.auto_create_partitions,
            )
    except async_durable_write_errors() as exc:
        handle_durable_failure(exc, self.on_error, entry.flags.fail_closed)

scrub

scrub(
    object_type: str,
    object_id: str,
    *,
    include_targets: bool = True,
    since: datetime | None = None,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult

Erase the values the entries of one object hold.

Every non-null value in data.changes and data.payload of the object's entries becomes "[erased]" (field keys stay, null stays null) and object_label becomes NULL. data.context is left to scrub_actor. Runs in one transaction on engine together with an audit.scrubbed entry recording who scrubbed which object and how many rows, without any erased value; nothing is committed if either fails. With auto_create_partitions, the partitions of the current month are ensured first.

Parameters:

Name	Type	Description	Default
object_type	str	Stored object_type of the object.	required
object_id	str	Stored object_id of the object, as audit_trail.diff.object_id_for formats it.	required
include_targets	bool	Also erase entries whose target is the object, such as changes of its children.	True
since	datetime | None	Only entries created at or after this aware datetime, which also limits the partitions scanned. None for all.	None
scope_ids	Collection[str] | None	Only entries whose scope_id is one of these, such as the tenants of an admin. None for no restriction; an empty collection erases nothing, and an entry without a scope never matches a restriction. Recorded in the audit.scrubbed entry.	None

Returns:

Type	Description
ScrubResult	A ScrubResult: activity_rows is the number of entries
ScrubResult	erased (entries already erased are not counted);
ScrubResult	transaction_rows is always 0.

Raises:

Type	Description
ScrubNotAllowedError	allow_scrub is False.
TypeError	engine is async; use ascrub; or scope_ids is a str.
ValueError	since is naive, or engine is in AUTOCOMMIT mode.

Source code in audit_trail/config.py

def scrub(
    self,
    object_type: str,
    object_id: str,
    *,
    include_targets: bool = True,
    since: datetime | None = None,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult:
    """Erase the values the entries of one object hold.

    Every non-null value in ``data.changes`` and ``data.payload`` of the
    object's entries becomes ``"[erased]"`` (field keys stay, ``null``
    stays ``null``) and ``object_label`` becomes ``NULL``. ``data.context``
    is left to ``scrub_actor``. Runs in one transaction on ``engine``
    together with an ``audit.scrubbed`` entry recording who scrubbed
    which object and how many rows, without any erased value; nothing is
    committed if either fails. With ``auto_create_partitions``, the
    partitions of the current month are ensured first.

    Args:
        object_type: Stored ``object_type`` of the object.
        object_id: Stored ``object_id`` of the object, as
            ``audit_trail.diff.object_id_for`` formats it.
        include_targets: Also erase entries whose target is the object,
            such as changes of its children.
        since: Only entries created at or after this aware datetime,
            which also limits the partitions scanned. ``None`` for all.
        scope_ids: Only entries whose ``scope_id`` is one of these, such
            as the tenants of an admin. ``None`` for no restriction; an
            empty collection erases nothing, and an entry without a
            scope never matches a restriction. Recorded in the
            ``audit.scrubbed`` entry.

    Returns:
        A ``ScrubResult``: ``activity_rows`` is the number of entries
        erased (entries already erased are not counted);
        ``transaction_rows`` is always ``0``.

    Raises:
        ScrubNotAllowedError: ``allow_scrub`` is ``False``.
        TypeError: ``engine`` is async; use ``ascrub``; or
            ``scope_ids`` is a ``str``.
        ValueError: ``since`` is naive, or ``engine`` is in
            ``AUTOCOMMIT`` mode.
    """
    from audit_trail import privacy

    privacy.check_allowed(self.allow_scrub)
    engine = self.engine
    if not isinstance(engine, Engine):
        raise TypeError("the engine is async; use ascrub")
    privacy.check_since(since)
    privacy.check_scope_ids(scope_ids)
    record = self._scrubbed_record()
    if self.auto_create_partitions:
        self.maintenance.ensure_partitions(months_ahead=0)
    with engine.begin() as conn:
        return privacy.scrub(
            conn,
            self.tables,
            object_type,
            object_id,
            include_targets=include_targets,
            since=since,
            scope_ids=scope_ids,
            record=record,
        )

ascrub async

ascrub(
    object_type: str,
    object_id: str,
    *,
    include_targets: bool = True,
    since: datetime | None = None,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult

Async scrub, for an AuditTrail built with an AsyncEngine.

Parameters:

Name	Type	Description	Default
object_type	str	Stored object_type of the object.	required
object_id	str	Stored object_id of the object.	required
include_targets	bool	Also erase entries whose target is the object.	True
since	datetime | None	Only entries created at or after this aware datetime.	None
scope_ids	Collection[str] | None	Only entries with one of these scope_id values; see scrub.	None

Returns:

Type	Description
ScrubResult	A ScrubResult; see scrub.

Raises:

Type	Description
ScrubNotAllowedError	allow_scrub is False.
TypeError	engine is sync; use scrub; or scope_ids is a str.
ValueError	As for scrub.

Source code in audit_trail/config.py

async def ascrub(
    self,
    object_type: str,
    object_id: str,
    *,
    include_targets: bool = True,
    since: datetime | None = None,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult:
    """Async ``scrub``, for an ``AuditTrail`` built with an ``AsyncEngine``.

    Args:
        object_type: Stored ``object_type`` of the object.
        object_id: Stored ``object_id`` of the object.
        include_targets: Also erase entries whose target is the object.
        since: Only entries created at or after this aware datetime.
        scope_ids: Only entries with one of these ``scope_id`` values;
            see ``scrub``.

    Returns:
        A ``ScrubResult``; see ``scrub``.

    Raises:
        ScrubNotAllowedError: ``allow_scrub`` is ``False``.
        TypeError: ``engine`` is sync; use ``scrub``; or ``scope_ids``
            is a ``str``.
        ValueError: As for ``scrub``.
    """
    from audit_trail import privacy

    privacy.check_allowed(self.allow_scrub)
    engine = self.engine
    if isinstance(engine, Engine):
        raise TypeError("the engine is sync; use scrub")
    privacy.check_since(since)
    privacy.check_scope_ids(scope_ids)
    record = self._scrubbed_record()
    if self.auto_create_partitions:
        await self.maintenance.aensure_partitions(months_ahead=0)
    async with engine.begin() as conn:
        return await conn.run_sync(
            lambda sync_conn: privacy.scrub(
                sync_conn,
                self.tables,
                object_type,
                object_id,
                include_targets=include_targets,
                since=since,
                scope_ids=scope_ids,
                record=record,
            )
        )

scrub_actor

scrub_actor(
    actor_id: str,
    *,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult

Clear the personal context stored for one actor.

audit_transaction rows created with this actor_id get actor_label, remote_addr, user_agent and meta set to NULL, and entries with this actor_id lose those keys from data.context. actor_id stays. A transaction row keeps the actor it was created with: a request that wrote its first entry anonymously and called set_actor afterwards keeps its address and user agent on that row. Runs in one transaction on engine with an audit.scrubbed entry, as scrub does.

With scope_ids, only entries whose scope_id is one of them lose the keys, and a transaction row is cleared only when it has entries and all of them, whoever their actor, are in those scopes: its fields are shared by all its entries, so a request that also wrote entries of another scope or without one keeps them, as does a transaction row without entries. Clearing an actor completely across scopes takes a run without scope_ids.

Parameters:

Name	Type	Description	Default
actor_id	str	The actor's actor_id.	required
scope_ids	Collection[str] | None	The scopes the scrub is restricted to, such as the tenants of an admin. None for no restriction; an empty collection changes nothing. Recorded in the audit.scrubbed entry.	None

Returns:

Type	Description
ScrubResult	A ScrubResult with the numbers of activity and transaction
ScrubResult	rows changed.

Raises:

Type	Description
ScrubNotAllowedError	allow_scrub is False.
TypeError	engine is async; use ascrub_actor; or scope_ids is a str.
ValueError	engine is in AUTOCOMMIT mode.

Source code in audit_trail/config.py

def scrub_actor(
    self, actor_id: str, *, scope_ids: Collection[str] | None = None
) -> ScrubResult:
    """Clear the personal context stored for one actor.

    ``audit_transaction`` rows created with this ``actor_id`` get
    ``actor_label``, ``remote_addr``, ``user_agent`` and ``meta`` set to
    ``NULL``, and entries with this ``actor_id`` lose those keys from
    ``data.context``. ``actor_id`` stays. A transaction row keeps the
    actor it was created with: a request that wrote its first entry
    anonymously and called ``set_actor`` afterwards keeps its address
    and user agent on that row. Runs in one transaction on ``engine``
    with an ``audit.scrubbed`` entry, as ``scrub`` does.

    With ``scope_ids``, only entries whose ``scope_id`` is one of them
    lose the keys, and a transaction row is cleared only when it has
    entries and all of them, whoever their actor, are in those scopes:
    its fields are shared by all its entries, so a request that also
    wrote entries of another scope or without one keeps them, as does a
    transaction row without entries. Clearing an actor completely
    across scopes takes a run without ``scope_ids``.

    Args:
        actor_id: The actor's ``actor_id``.
        scope_ids: The scopes the scrub is restricted to, such as the
            tenants of an admin. ``None`` for no restriction; an empty
            collection changes nothing. Recorded in the
            ``audit.scrubbed`` entry.

    Returns:
        A ``ScrubResult`` with the numbers of activity and transaction
        rows changed.

    Raises:
        ScrubNotAllowedError: ``allow_scrub`` is ``False``.
        TypeError: ``engine`` is async; use ``ascrub_actor``; or
            ``scope_ids`` is a ``str``.
        ValueError: ``engine`` is in ``AUTOCOMMIT`` mode.
    """
    from audit_trail import privacy

    privacy.check_allowed(self.allow_scrub)
    engine = self.engine
    if not isinstance(engine, Engine):
        raise TypeError("the engine is async; use ascrub_actor")
    privacy.check_scope_ids(scope_ids)
    record = self._scrubbed_record()
    if self.auto_create_partitions:
        self.maintenance.ensure_partitions(months_ahead=0)
    with engine.begin() as conn:
        return privacy.scrub_actor(
            conn, self.tables, actor_id, scope_ids=scope_ids, record=record
        )

ascrub_actor async

ascrub_actor(
    actor_id: str,
    *,
    scope_ids: Collection[str] | None = None,
) -> ScrubResult

Async scrub_actor, for an AuditTrail built with an AsyncEngine.

Parameters:

Name	Type	Description	Default
actor_id	str	The actor's actor_id.	required
scope_ids	Collection[str] | None	The scopes the scrub is restricted to; see scrub_actor.	None

Returns:

Type	Description
ScrubResult	A ScrubResult with the numbers of activity and transaction
ScrubResult	rows changed.

Raises:

Type	Description
ScrubNotAllowedError	allow_scrub is False.
TypeError	engine is sync; use scrub_actor; or scope_ids is a str.
ValueError	engine is in AUTOCOMMIT mode.

Source code in audit_trail/config.py

async def ascrub_actor(
    self, actor_id: str, *, scope_ids: Collection[str] | None = None
) -> ScrubResult:
    """Async ``scrub_actor``, for an ``AuditTrail`` built with an ``AsyncEngine``.

    Args:
        actor_id: The actor's ``actor_id``.
        scope_ids: The scopes the scrub is restricted to; see
            ``scrub_actor``.

    Returns:
        A ``ScrubResult`` with the numbers of activity and transaction
        rows changed.

    Raises:
        ScrubNotAllowedError: ``allow_scrub`` is ``False``.
        TypeError: ``engine`` is sync; use ``scrub_actor``; or
            ``scope_ids`` is a ``str``.
        ValueError: ``engine`` is in ``AUTOCOMMIT`` mode.
    """
    from audit_trail import privacy

    privacy.check_allowed(self.allow_scrub)
    engine = self.engine
    if isinstance(engine, Engine):
        raise TypeError("the engine is sync; use scrub_actor")
    privacy.check_scope_ids(scope_ids)
    record = self._scrubbed_record()
    if self.auto_create_partitions:
        await self.maintenance.aensure_partitions(months_ahead=0)
    async with engine.begin() as conn:
        return await conn.run_sync(
            lambda sync_conn: privacy.scrub_actor(
                sync_conn,
                self.tables,
                actor_id,
                scope_ids=scope_ids,
                record=record,
            )
        )