diff --git a/CHANGELOG.md b/CHANGELOG.md index a11856d..76e7f8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ - `log()` / `alog()` accept just the event and take the session from `AuditTrail(session_provider=...)`; a missing session or one of the wrong kind raises a clear error - `AuditTrail(default_query_window=...)`: `list_groups()` without `since` reads only that window; `since=ALL_HISTORY` reads the whole log, the cursor keeps the first page's bound, and `Page.since` tells where the window started - Under `AsyncSession`, loading an expired column of an `Audited` model (for example assigning to it after a commit) or an unloaded tracked collection raises `AsyncLoadError`, a `MissingGreenlet` subclass that names the attribute and the fix +- Sessions without a default bind work: audit rows go on the connection the flush used for the object (per mapper or table `binds`, or an overridden `get_bind`), one `audit_transaction` row per database; `log()` without `obj` and the query API use a bind of the audit tables, otherwise raise a clear `UnboundExecutionError` ## 0.1.0a1 - 2026-09-25 diff --git a/audit_trail/config.py b/audit_trail/config.py index 2c67985..c20f9ba 100644 --- a/audit_trail/config.py +++ b/audit_trail/config.py @@ -367,11 +367,13 @@ def log( ``None``), the entry goes to the session ``session_provider()`` returns. - A non-durable entry is inserted immediately on - ``session.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``. + 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 @@ -430,10 +432,13 @@ def log( ``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 write_entries + from audit_trail.listener import PendingEntry, write_entries from audit_trail.writer import ( DURABLE_WRITE_ERRORS, handle_durable_failure, @@ -449,7 +454,7 @@ def log( ) entry = self._prepare(session, call.event, obj, target, payload, actor, durable) if not entry.flags.durable: - write_entries(session, self, [entry.entry], entry.ctx) + write_entries(session, self, [PendingEntry(entry.entry, obj)], entry.ctx) return engine = self.durable_engine if not isinstance(engine, Engine): @@ -543,10 +548,11 @@ async def alog( 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 write_entries + from audit_trail.listener import PendingEntry, write_entries from audit_trail.writer import ( DURABLE_WRITE_ERRORS, handle_durable_failure, @@ -566,7 +572,7 @@ async def alog( if not entry.flags.durable: await session.run_sync( lambda sync_session: write_entries( - sync_session, self, [entry.entry], entry.ctx + sync_session, self, [PendingEntry(entry.entry, obj)], entry.ctx ) ) return diff --git a/audit_trail/listener.py b/audit_trail/listener.py index fdaf055..128f302 100644 --- a/audit_trail/listener.py +++ b/audit_trail/listener.py @@ -8,16 +8,18 @@ is installed through that class, which must be the host's own subclass. ``after_flush`` turns the flushed ``Audited`` instances into ``entity.*`` -entries and writes them with plain SQL on ``session.connection()``, in the -same database transaction as the changes, so a rollback undoes both. +entries and writes them with plain SQL on the session's connection for each +object (resolved like the flush does), in the same database transaction as +the changes, so a rollback undoes both. The ``audit_transaction`` row is inserted by the first entry of a database -transaction and cached in ``session.info`` together with the -``SessionTransaction`` (root or savepoint) it was inserted in: +transaction on each connection and cached in ``session.info``, per +connection, together with the ``SessionTransaction`` (root or savepoint) it +was inserted in: - rolling back that ``SessionTransaction`` or one of its ancestors - (``after_soft_rollback``) drops the cache, because the row is gone; -- ``after_transaction_end`` drops it when the outermost transaction ends, + (``after_soft_rollback``) drops the cached row, because it is gone; +- ``after_transaction_end`` drops the cache when the outermost transaction ends, however it ends: commit, rollback, or ``Session.close()`` (which fires no commit or rollback event). Releasing a savepoint also ends a ``SessionTransaction``, but the row then lives on in the enclosing one; @@ -75,6 +77,7 @@ pop_relationship_changes, track_relationships, ) +from audit_trail.tables import audit_connection from audit_trail.writer import ( ENVELOPE_VERSION, Entry, @@ -87,7 +90,7 @@ ) if TYPE_CHECKING: - from sqlalchemy import Result + from sqlalchemy import Connection, Result from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.orm import ( InstanceState, @@ -142,6 +145,13 @@ class _CachedRow(NamedTuple): row: TransactionRow +class PendingEntry(NamedTuple): + """An entry to write and the object it is about, if any.""" + + entry: Entry + obj: object | None + + def install( trail: AuditTrail, session_factory: type[Session | AsyncSession] @@ -337,7 +347,7 @@ def do_orm_execute(self, orm_execute_state: ORMExecuteState) -> None: model.__qualname__, ) - def _entries(self, session: Session, ctx: AuditContext) -> list[Entry]: + def _entries(self, session: Session, ctx: AuditContext) -> list[PendingEntry]: trail = self.trail context = context_data(ctx, trail.json_encoder) batches: tuple[tuple[ChangeKind, Iterable[object]], ...] = ( @@ -345,14 +355,14 @@ def _entries(self, session: Session, ctx: AuditContext) -> list[Entry]: ("updated", session.dirty), ("deleted", session.deleted), ) - entries: list[Entry] = [] + entries: list[PendingEntry] = [] for kind, objects in batches: for obj in objects: if not isinstance(obj, Audited): continue entry = self._entry(obj, kind, ctx, context) if entry is not None: - entries.append(entry) + entries.append(PendingEntry(entry, obj)) # Only now: the snapshots were the old values of this flush's changes. for obj in chain(session.new, session.dirty): if isinstance(obj, Audited): @@ -517,22 +527,73 @@ def _audited_model(state: ORMExecuteState) -> type[Any] | None: def write_entries( - session: Session, trail: AuditTrail, entries: Sequence[Entry], ctx: AuditContext + session: Session, + trail: AuditTrail, + entries: Sequence[PendingEntry], + ctx: AuditContext, ) -> None: """Write entries in the session's current database transaction. - Inserts the ``audit_transaction`` row first when the transaction has - none yet, then all entries with one ``executemany``, following - ``trail.on_error``. Uses only ``session.connection()``, never the ORM. + Each entry goes on the connection its object's rows were flushed on: + ``session.connection()`` with the base mapper of the object, as the flush + resolves it (mapper binds, then its table's, then the default bind). An + entry without an object uses the audit tables as the clause (their binds, + then the default bind). A ``get_bind`` override is honoured. + + Per connection, inserts the ``audit_transaction`` row first when that + connection's transaction has none yet, then its entries with one + ``executemany``, following ``trail.on_error``. Uses only the connections, + never the ORM. Args: session: A session of an installed class. trail: The configuration to write with. - entries: The entries. - ctx: The context, used when the transaction row is created. + entries: The entries, with the objects they are about. + ctx: The context, used when a transaction row is created. + + Raises: + UnboundExecutionError: An entry without an object, and the session + has no bind for the audit tables. """ - connection = session.connection() - cached = _cached_row(session) + by_mapper: dict[Mapper[Any] | None, Connection] = {} + groups: dict[Connection, list[Entry]] = {} + for pending in entries: + mapper = _base_mapper(pending.obj) + connection = by_mapper.get(mapper) + if connection is None: + connection = by_mapper[mapper] = _connection_for(session, trail, mapper) + groups.setdefault(connection, []).append(pending.entry) + for connection, group in groups.items(): + _write_group(session, trail, connection, group, ctx) + + +def _base_mapper(obj: object | None) -> Mapper[Any] | None: + # The flush resolves an object's connection from its base mapper. + state = None if obj is None else inspect(obj, raiseerr=False) + mapper: Mapper[Any] | None = getattr(state, "mapper", None) + return None if mapper is None else mapper.base_mapper + + +def _connection_for( + session: Session, trail: AuditTrail, mapper: Mapper[Any] | None +) -> Connection: + if mapper is not None: + return session.connection(bind_arguments={"mapper": mapper}) + return audit_connection( + session, + trail.tables, + " log() and alog() with obj= use the bind of the object instead.", + ) + + +def _write_group( + session: Session, + trail: AuditTrail, + connection: Connection, + entries: Sequence[Entry], + ctx: AuditContext, +) -> None: + cached = _cached_row(session, connection) def work() -> TransactionRow: row = cached or insert_transaction( @@ -547,11 +608,13 @@ def work() -> TransactionRow: if row is not None and cached is None: owner = session.get_nested_transaction() or session.get_transaction() assert owner is not None # session.connection() began one - session.info[_CACHE_KEY] = _CachedRow(owner, row) + rows: dict[Connection, _CachedRow] = session.info.setdefault(_CACHE_KEY, {}) + rows[connection] = _CachedRow(owner, row) -def _cached_row(session: Session) -> TransactionRow | None: - cached: _CachedRow | None = session.info.get(_CACHE_KEY) +def _cached_row(session: Session, connection: Connection) -> TransactionRow | None: + rows: dict[Connection, _CachedRow] = session.info.get(_CACHE_KEY, {}) + cached = rows.get(connection) return None if cached is None else cached.row @@ -559,7 +622,7 @@ def _after_transaction_end(session: Session, transaction: SessionTransaction) -> # Fires for every way a transaction ends, including Session.close() # without commit or rollback, which fires no commit or rollback event. # A released savepoint or a flush's subtransaction is not the end of the - # database transaction: the row lives on in the enclosing one. + # database transaction: the rows live on in the enclosing one. if transaction.parent is None: session.info.pop(_CACHE_KEY, None) @@ -567,15 +630,23 @@ def _after_transaction_end(session: Session, transaction: SessionTransaction) -> def _after_soft_rollback( session: Session, previous_transaction: SessionTransaction ) -> None: - cached: _CachedRow | None = session.info.get(_CACHE_KEY) - if cached is None: + rows: dict[Connection, _CachedRow] | None = session.info.get(_CACHE_KEY) + if not rows: return - transaction: SessionTransaction | None = cached.owner - while transaction is not None: - if transaction is previous_transaction: - session.info.pop(_CACHE_KEY, None) - return - transaction = transaction.parent + # A savepoint spans every connection of the session, so its rollback + # removes the rows it holds on each. + for connection, cached in list(rows.items()): + if _within(cached.owner, previous_transaction): + del rows[connection] + + +def _within(transaction: SessionTransaction, ancestor: SessionTransaction) -> bool: + current: SessionTransaction | None = transaction + while current is not None: + if current is ancestor: + return True + current = current.parent + return False def _after_rollback(session: Session) -> None: diff --git a/audit_trail/query.py b/audit_trail/query.py index 07b0c40..e68d06b 100644 --- a/audit_trail/query.py +++ b/audit_trail/query.py @@ -22,6 +22,9 @@ Every query runs 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. +The setting and the query run on one connection of the session: the one for +the audit tables, that is their bind in ``Session(binds=...)``, else the +session's default bind; with neither, ``UnboundExecutionError`` is raised. """ from __future__ import annotations @@ -59,10 +62,10 @@ from audit_trail.checks import _registry_of from audit_trail.diff import REDACTED, UNKNOWN, field_policy, options_of from audit_trail.serialization import JSONValue -from audit_trail.tables import AuditTables +from audit_trail.tables import AuditTables, audit_connection if TYPE_CHECKING: - from sqlalchemy import ColumnClause, Table + from sqlalchemy import ColumnClause, Connection, Table from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapper, Session, registry @@ -466,10 +469,11 @@ def list_groups( reports the bound: pass it as ``until`` to read the window before. 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. + needed), on its connection for the audit tables, 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. @@ -506,6 +510,8 @@ def list_groups( datetime of the cursor is not timezone-aware. TypeError: A collection filter is a ``str``, or ``since`` is neither a datetime, ``ALL_HISTORY`` nor ``None``. + UnboundExecutionError: The session has neither a bind for the + audit tables nor a default bind. """ _check_page(since, until, cursor, limit) bound = self._bound(since, until, cursor, self.default_window) @@ -658,13 +664,14 @@ def get( conditions.append(a.c.severity == severity) if visibility is not None: conditions.append(visibility.predicate(a)) - session.execute(_CUSTOM_PLAN) - row = session.execute(select(a).where(*conditions)).mappings().first() + connection = audit_connection(session, self.tables) + connection.execute(_CUSTOM_PLAN) + row = connection.execute(select(a).where(*conditions)).mappings().first() if row is None: return None activity = _activity(row) header = ( - session.execute( + connection.execute( select(t).where( t.c.id == activity["transaction_id"], t.c.issued_at == activity["created_at"], @@ -977,8 +984,9 @@ def access_summary( .group_by(a.c.actor_id) .order_by(last_at.desc(), a.c.actor_id) ) - session.execute(_CUSTOM_PLAN) - return [AccessCount(*row) for row in session.execute(statement)] + connection = audit_connection(session, self.tables) + connection.execute(_CUSTOM_PLAN) + return [AccessCount(*row) for row in connection.execute(statement)] async def aaccess_summary( self, @@ -1074,11 +1082,12 @@ def _page( # matches the rows of all streams together; `scope` narrows the rows # a group shows (None: every visible row of the transaction); `bound` # is recorded in the next cursor. - session.execute(_CUSTOM_PLAN) - shown = self._shown(session, boundary, cursor) - selection = self._select(session, streams, cursor, shown, limit) + connection = audit_connection(session, self.tables) + connection.execute(_CUSTOM_PLAN) + shown = self._shown(connection, boundary, cursor) + selection = self._select(connection, streams, cursor, shown, limit) groups = self._groups( - session, selection.transactions, visibility, compact, scope + connection, selection.transactions, visibility, compact, scope ) next_cursor = selection.next_cursor if next_cursor is not None: @@ -1087,7 +1096,7 @@ def _page( def _shown( self, - session: Session, + connection: Connection, conditions: list[ColumnElement[bool]], cursor: Cursor | None, ) -> frozenset[int]: @@ -1106,18 +1115,18 @@ def _shown( a.c.id >= cursor.id, ) ) - return frozenset(session.scalars(statement)) + return frozenset(connection.scalars(statement)) def _select( self, - session: Session, + connection: Connection, streams: list[list[ColumnElement[bool]]], cursor: Cursor | None, shown: frozenset[int], limit: int, ) -> _Selection: merged = heapq.merge( - *(self._hits(session, stream, cursor, limit + 1) for stream in streams), + *(self._hits(connection, stream, cursor, limit + 1) for stream in streams), key=lambda hit: (hit.created_at, hit.id), reverse=True, ) @@ -1140,7 +1149,7 @@ def _select( def _hits( self, - session: Session, + connection: Connection, conditions: list[ColumnElement[bool]], cursor: Cursor | None, batch: int, @@ -1163,7 +1172,7 @@ def _hits( .order_by(a.c.created_at.desc(), a.c.id.desc()) .limit(batch) ) - hits = [_Hit(*row) for row in session.execute(statement)] + hits = [_Hit(*row) for row in connection.execute(statement)] yield from hits if len(hits) < batch: return @@ -1171,7 +1180,7 @@ def _hits( def _groups( self, - session: Session, + connection: Connection, transactions: dict[int, datetime], visibility: Visibility | None, compact: bool, @@ -1195,12 +1204,12 @@ def _groups( conditions.append(scope) rows: dict[int, list[ActivityRow]] = {} statement = select(a).where(*conditions).order_by(a.c.transaction_id, a.c.id) - for mapping in session.execute(statement).mappings(): + for mapping in connection.execute(statement).mappings(): activity = _activity(mapping) rows.setdefault(activity["transaction_id"], []).append(activity) headers = { mapping["id"]: mapping - for mapping in session.execute( + for mapping in connection.execute( select(t).where(t.c.id.in_(ids), t.c.issued_at.between(low, high)) ).mappings() } @@ -1403,10 +1412,11 @@ def _labels( latest.c.id, latest.c.object_label, ).select_from(references.join(latest, true())) - session.execute(_CUSTOM_PLAN) + connection = audit_connection(session, self.tables) + connection.execute(_CUSTOM_PLAN) return { _Reference(object_type, object_id): _Label(created_at, row_id, label) - for object_type, object_id, created_at, row_id, label in session.execute( + for object_type, object_id, created_at, row_id, label in connection.execute( statement ) } diff --git a/audit_trail/tables.py b/audit_trail/tables.py index f325a4b..37291b0 100644 --- a/audit_trail/tables.py +++ b/audit_trail/tables.py @@ -18,7 +18,7 @@ from collections.abc import Collection from dataclasses import dataclass -from typing import Literal, get_args +from typing import TYPE_CHECKING, Literal, get_args from sqlalchemy import ( BigInteger, @@ -35,9 +35,14 @@ text, ) from sqlalchemy.dialects.postgresql import INET, JSONB, UUID +from sqlalchemy.exc import UnboundExecutionError from audit_trail._typing import assert_never +if TYPE_CHECKING: + from sqlalchemy import Connection + from sqlalchemy.orm import Session + MAX_IDENTIFIER_LENGTH = 63 """PostgreSQL's limit on identifier length in bytes; longer names are truncated.""" @@ -83,6 +88,33 @@ class AuditTables: activity: Table +def audit_connection( + session: Session, tables: AuditTables, hint: str = "" +) -> Connection: + """The session's connection for the audit tables. + + Resolved by ``Session.get_bind`` with the ``audit_activity`` table as the + clause: a bind of that table in ``Session(binds=...)``, else the default + bind. A ``get_bind`` override is honoured. + + Args: + session: The session. + tables: The audit tables. + hint: Appended to the error message. + + Raises: + UnboundExecutionError: Neither resolves, chained to SQLAlchemy's error. + """ + try: + return session.connection(bind_arguments={"clause": tables.activity}) + except UnboundExecutionError as exc: + raise UnboundExecutionError( + "The session has no bind for the audit tables: bind " + f"{tables.activity.fullname} and {tables.transaction.fullname} in " + f"Session(binds=...), or give the session a default bind.{hint}" + ) from exc + + def build_tables( *, schema: str = "audit", diff --git a/docs/guides/models.md b/docs/guides/models.md index a7912e8..5285604 100644 --- a/docs/guides/models.md +++ b/docs/guides/models.md @@ -154,3 +154,26 @@ def test_audited_models() -> None: | `json-in-place` | warning | a JSON column is neither `Mutable` nor in `snapshot_on_load` | A column whose name only looks sensitive can be allowed with `check_models(Base, allow_names={"Invoice.token_count"})`. + +## Sessions with binds + +A session without a default bind, configured with `binds` per model or table, works as SQLAlchemy resolves it (`Session.get_bind`, including an override of it in a `Session` subclass): + +- An `entity.*` entry, and a `log(obj=...)` entry, go on the connection the object is flushed on, resolved from its base mapper: in the same database transaction as the object's rows. +- `log()` without `obj`, and the queries of `trail.query`, use the connection for the audit tables: their bind in `binds`, else the session's default bind. With neither, they raise `UnboundExecutionError`. + +```python +from sqlalchemy.orm import sessionmaker + +tables = trail.tables +Session = sessionmaker( + binds={ + Base: engine, + tables.activity: engine, # for log() without obj and trail.query + tables.transaction: engine, + } +) +trail.install(Session) +``` + +When one flush writes objects bound to different databases, each database gets its entries and an `audit_transaction` row of its own, in its transaction there, so one commit has one `transaction_id` per database. The audit tables must exist in each of them. Horizontal sharding (`ShardedSession`) is not supported. diff --git a/docs/limitations.md b/docs/limitations.md index 4a72f72..6654540 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -12,6 +12,10 @@ The library records what passes through SQLAlchemy's unit of work, in Python. Th Statements run on a plain `Connection` or another engine are not seen at all. +## Sessions with binds + +A session without a default bind needs the audit tables in its `binds` for `log()` without `obj` and for queries; see [Sessions with binds](guides/models.md#sessions-with-binds). + ## Database-side cascades Rows deleted by the database, through `ForeignKey(..., ondelete="CASCADE")` with `passive_deletes=True`, never pass through the session, so they get no `entity.deleted` entry. The same goes for triggers and anything else the database changes on its own. Let SQLAlchemy cascade the delete (`cascade="all, delete-orphan"` without `passive_deletes`) for models whose deletion must be audited. diff --git a/tests/db/test_binds.py b/tests/db/test_binds.py new file mode 100644 index 0000000..6db65c5 --- /dev/null +++ b/tests/db/test_binds.py @@ -0,0 +1,392 @@ +"""Sessions without a default bind, configured with ``binds``. + +Entity entries and ``log(obj=...)`` go on the connection the object is flushed +on; ``log()`` without an object and the queries on the connection for the +audit tables. Most tests run once per session kind (see ``listener_support``); +a second bind is a second engine on the test database. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable, Mapping +from dataclasses import dataclass +from typing import Any + +import pytest +from sqlalchemy import Engine, ForeignKey, MetaData, RowMapping, create_engine, select +from sqlalchemy.exc import UnboundExecutionError +from sqlalchemy.ext.asyncio import ( + AsyncEngine, + AsyncSession, + async_sessionmaker, + create_async_engine, +) +from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker + +from audit_trail import Audited, AuditEvent, AuditTrail, Severity, event +from tests.db.listener_support import ( + SESSION_KINDS, + SessionKind, + create_trail, + each_session_kind, + session_kind, +) + +NO_AUDIT_BIND = "no bind for the audit tables" + +Bind = Engine | AsyncEngine + + +class BindEvent(AuditEvent): + SEEN = event("binds_test.seen", Severity.NOTICE) + + +@dataclass +class Models: + Base: Any + Doc: Any + Page: Any + OtherBase: Any + Note: Any + + +@pytest.fixture +def models(engine: Engine, schema: str) -> Models: + class Base(DeclarativeBase): + metadata = MetaData(schema=schema) + + class Doc(Base, Audited): + __tablename__ = "doc" + id: Mapped[int] = mapped_column(primary_key=True) + kind: Mapped[str] = mapped_column(default="doc") + + __mapper_args__ = { # noqa: RUF012 + "polymorphic_on": "kind", + "polymorphic_identity": "doc", + } + + class Page(Doc): + __tablename__ = "page" + id: Mapped[int] = mapped_column(ForeignKey("doc.id"), primary_key=True) + + __mapper_args__ = {"polymorphic_identity": "page"} # noqa: RUF012 + + class OtherBase(DeclarativeBase): + metadata = MetaData(schema=schema) + + class Note(OtherBase, Audited): + __tablename__ = "note" + id: Mapped[int] = mapped_column(primary_key=True) + + Base.metadata.create_all(engine) + OtherBase.metadata.create_all(engine) + return Models(Base, Doc, Page, OtherBase, Note) + + +@pytest.fixture(params=SESSION_KINDS) +async def kind( + request: pytest.FixtureRequest, engine: Engine +) -> AsyncIterator[SessionKind]: + async with session_kind(request.param, engine) as value: + yield value + + +@pytest.fixture +def trail(engine: Engine, schema: str, kind: SessionKind) -> AuditTrail: + return create_trail( + engine, + schema, + trail_engine=kind.async_engine, + severities=Severity, + events=[BindEvent], + ) + + +@pytest.fixture +def first(engine: Engine, kind: SessionKind) -> Bind: + """The engine of the session kind.""" + return engine if kind.async_engine is None else kind.async_engine + + +@pytest.fixture +async def second(engine: Engine, kind: SessionKind) -> AsyncIterator[Bind]: + """Another engine of the session kind, on the test database.""" + if kind.async_engine is None: + other_sync = create_engine(engine.url) + yield other_sync + other_sync.dispose() + return + other = create_async_engine(kind.async_engine.url) + yield other + await other.dispose() + + +def sessions( + trail: AuditTrail, kind: SessionKind, binds: Mapping[Any, Bind] +) -> Callable[[], Session]: + """Sessions of a new installed factory with ``binds`` and no default bind.""" + if kind.async_engine is None: + factory = sessionmaker(binds=dict(binds)) + trail.install(factory) + return factory + async_factory = async_sessionmaker( + binds=dict(binds), sync_session_class=type("BindsSession", (Session,), {}) + ) + trail.install(async_factory) + return lambda: async_factory().sync_session + + +def audit_binds(trail: AuditTrail, bind: Bind) -> dict[Any, Bind]: + return {trail.tables.activity: bind, trail.tables.transaction: bind} + + +def activities(engine: Engine, trail: AuditTrail) -> list[RowMapping]: + table = trail.tables.activity + with engine.connect() as conn: + return list(conn.execute(select(table).order_by(table.c.id)).mappings()) + + +def transaction_ids(engine: Engine, trail: AuditTrail) -> list[int]: + table = trail.tables.transaction + with engine.connect() as conn: + return list(conn.scalars(select(table.c.id).order_by(table.c.id))) + + +@each_session_kind +def test_mapper_bind_is_used_for_capture_and_log( + engine: Engine, trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Base: first}) + with factory() as session: + doc = models.Doc(id=1) + session.add(doc) + session.flush() + trail.log(session, BindEvent.SEEN, obj=doc) + session.commit() + + rows = activities(engine, trail) + assert [row["verb"] for row in rows] == ["entity.created", "binds_test.seen"] + assert len(transaction_ids(engine, trail)) == 1 + assert {row["transaction_id"] for row in rows} == set( + transaction_ids(engine, trail) + ) + + +@each_session_kind +def test_table_bind_of_the_model_is_used( + engine: Engine, trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Note.__table__: first}) + with factory() as session: + session.add(models.Note(id=1)) + session.commit() + + assert [row["verb"] for row in activities(engine, trail)] == ["entity.created"] + + +@each_session_kind +def test_log_without_obj_uses_the_bind_of_the_audit_tables( + engine: Engine, trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Base: first, **audit_binds(trail, first)}) + with factory() as session: + session.add(models.Doc(id=1)) + session.flush() + trail.log(session, BindEvent.SEEN) + session.commit() + + rows = activities(engine, trail) + assert [row["verb"] for row in rows] == ["entity.created", "binds_test.seen"] + # One engine: one connection, so one transaction row. + assert len(transaction_ids(engine, trail)) == 1 + + +@each_session_kind +def test_log_without_obj_and_without_an_audit_bind_raises( + engine: Engine, trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Base: first}) + with factory() as session: + with pytest.raises(UnboundExecutionError, match=NO_AUDIT_BIND) as raised: + trail.log(session, BindEvent.SEEN) + assert isinstance(raised.value.__cause__, UnboundExecutionError) + + assert activities(engine, trail) == [] + + +@each_session_kind +def test_one_transaction_row_per_bind( + engine: Engine, + trail: AuditTrail, + models: Models, + kind: SessionKind, + first: Bind, + second: Bind, +) -> None: + factory = sessions(trail, kind, {models.Base: first, models.OtherBase: second}) + with factory() as session: + session.add_all([models.Doc(id=1), models.Note(id=1)]) + session.flush() + session.add_all([models.Doc(id=2), models.Note(id=2)]) + session.commit() + + rows = activities(engine, trail) + by_type: dict[str, set[int]] = {} + for row in rows: + by_type.setdefault(row["object_type"], set()).add(row["transaction_id"]) + assert len(rows) == 4 + assert len(by_type["Doc"]) == 1 + assert len(by_type["Note"]) == 1 + assert by_type["Doc"] != by_type["Note"] + assert by_type["Doc"] | by_type["Note"] == set(transaction_ids(engine, trail)) + + +@each_session_kind +def test_savepoint_rollback_drops_the_row_of_every_bind( + engine: Engine, + trail: AuditTrail, + models: Models, + kind: SessionKind, + first: Bind, + second: Bind, +) -> None: + factory = sessions(trail, kind, {models.Base: first, models.OtherBase: second}) + with factory() as session: + # Both connections join the session before the savepoint. + session.connection(bind_arguments={"mapper": models.Doc}) + session.connection(bind_arguments={"mapper": models.Note}) + savepoint = session.begin_nested() + session.add_all([models.Doc(id=1), models.Note(id=1)]) + session.flush() + savepoint.rollback() + session.add_all([models.Doc(id=2), models.Note(id=2)]) + session.commit() + + rows = activities(engine, trail) + assert [row["object_id"] for row in rows] == ["2", "2"] + # Every entry points at a transaction row that was committed. + assert {row["transaction_id"] for row in rows} == set( + transaction_ids(engine, trail) + ) + assert len(transaction_ids(engine, trail)) == 2 + + +@each_session_kind +def test_subclass_entry_uses_the_bind_of_the_base_mapper( + engine: Engine, + trail: AuditTrail, + models: Models, + kind: SessionKind, + first: Bind, + second: Bind, +) -> None: + # The flush writes a Page on the bind of its base mapper, Doc's; its + # entry follows it into that transaction. + factory = sessions(trail, kind, {models.Base: first, models.Page: second}) + with factory() as session: + session.add_all([models.Doc(id=1), models.Page(id=2)]) + session.commit() + + assert len(activities(engine, trail)) == 2 + assert len(transaction_ids(engine, trail)) == 1 + + +def test_get_bind_override_is_honoured( + engine: Engine, schema: str, models: Models +) -> None: + class RoutingSession(Session): + def get_bind(self, mapper: Any = None, clause: Any = None, **kw: Any) -> Any: + if mapper is None and clause is None: + raise UnboundExecutionError("no routing context") + return engine + + trail = create_trail(engine, schema, severities=Severity, events=[BindEvent]) + factory = sessionmaker(class_=RoutingSession) + trail.install(factory) + with factory() as session: + doc = models.Doc(id=1) + session.add(doc) + session.flush() + trail.log(session, BindEvent.SEEN, obj=doc) + trail.log(session, BindEvent.SEEN) + session.commit() + with factory() as session: + page = trail.query.list_groups(session) + + assert [len(group.activities) for group in page.groups] == [3] + assert len(transaction_ids(engine, trail)) == 1 + + +@each_session_kind +def test_list_groups_uses_the_bind_of_the_audit_tables( + trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Base: first, **audit_binds(trail, first)}) + with factory() as session: + session.add(models.Doc(id=1)) + session.commit() + with factory() as session: + page = trail.query.list_groups(session) + + assert [len(group.activities) for group in page.groups] == [1] + + +@each_session_kind +def test_list_groups_without_an_audit_bind_raises( + trail: AuditTrail, models: Models, kind: SessionKind, first: Bind +) -> None: + factory = sessions(trail, kind, {models.Base: first}) + with factory() as session: + with pytest.raises(UnboundExecutionError, match=NO_AUDIT_BIND) as raised: + trail.query.list_groups(session) + assert isinstance(raised.value.__cause__, UnboundExecutionError) + + +async def test_async_session_with_binds( + engine: Engine, schema: str, models: Models, async_engine: AsyncEngine +) -> None: + trail = create_trail( + engine, + schema, + trail_engine=async_engine, + severities=Severity, + events=[BindEvent], + ) + factory = async_sessionmaker( + binds={models.Base: async_engine, **audit_binds(trail, async_engine)}, + sync_session_class=type("BindsSession", (Session,), {}), + ) + trail.install(factory) + async with factory() as session: + doc = models.Doc(id=1) + session.add(doc) + await session.flush() + await trail.alog(session, BindEvent.SEEN, obj=doc) + await trail.alog(session, BindEvent.SEEN) + await session.commit() + async with factory() as session: + page = await trail.query.alist_groups(session) + + assert [len(group.activities) for group in page.groups] == [3] + + +async def test_async_session_without_an_audit_bind_raises( + engine: Engine, schema: str, models: Models, async_engine: AsyncEngine +) -> None: + trail = create_trail( + engine, + schema, + trail_engine=async_engine, + severities=Severity, + events=[BindEvent], + ) + factory: async_sessionmaker[AsyncSession] = async_sessionmaker( + binds={models.Base: async_engine}, + sync_session_class=type("BindsSession", (Session,), {}), + ) + trail.install(factory) + async with factory() as session: + with pytest.raises(UnboundExecutionError, match=NO_AUDIT_BIND): + await trail.query.alist_groups(session) + with pytest.raises(UnboundExecutionError, match=NO_AUDIT_BIND): + await trail.alog(session, BindEvent.SEEN)