diff --git a/CHANGELOG.md b/CHANGELOG.md index 03e7b00..a11856d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ - `AuditTrail.scrub()` / `scrub_actor()` (and async variants), enabled with `allow_scrub=True`: erase an object's stored values or an actor's personal context in one transaction, recorded as an `audit.scrubbed` entry; erased values are never merged away when entries are grouped - `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 ## 0.1.0a1 - 2026-09-25 diff --git a/audit_trail/__init__.py b/audit_trail/__init__.py index 429508f..7fdfa3b 100644 --- a/audit_trail/__init__.py +++ b/audit_trail/__init__.py @@ -7,6 +7,7 @@ from audit_trail.context import Actor, AuditContext from audit_trail.diff import AuditOptionError, FieldPolicyError from audit_trail.events import AuditEvent, Severity, event +from audit_trail.listener import AsyncLoadError from audit_trail.mixin import Audited from audit_trail.privacy import ScrubNotAllowedError, ScrubResult from audit_trail.query import ( @@ -37,6 +38,7 @@ "ActivityDetail", "Actor", "AllHistory", + "AsyncLoadError", "AuditContext", "AuditEvent", "AuditOptionError", diff --git a/audit_trail/listener.py b/audit_trail/listener.py index f49824a..fdaf055 100644 --- a/audit_trail/listener.py +++ b/audit_trail/listener.py @@ -28,6 +28,16 @@ ``UPDATE``/``DELETE`` statements on an audited table executed through the session: they bypass the flush and get no entry. +A second ``do_orm_execute`` handler explains loads that fail under +``AsyncSession``. Loading an unloaded attribute outside an awaited call needs +I/O that SQLAlchemy cannot run there, so it raises ``MissingGreenlet``. For +the loads the library is responsible for (an expired column of an +``Audited`` instance, which ``Audited`` also loads on assignment to keep the +old value, and a collection named in ``track_relationships``) the error is +replaced by ``AsyncLoadError``, which names the attribute and the fix. The +handler adds no SQL and checks nothing in advance: it only translates the +error once the load has failed, and only on an async dialect. + These state listeners run even when ``session.info["audit_enabled"]`` is ``False``; only capture is switched off, so toggling the flag in the middle of a transaction cannot leave a stale cache behind. @@ -42,6 +52,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast from sqlalchemy import event, inspect +from sqlalchemy.exc import MissingGreenlet, StatementError from sqlalchemy.orm import Session, sessionmaker from audit_trail._compaction import FieldChange @@ -76,8 +87,10 @@ ) if TYPE_CHECKING: + from sqlalchemy import Result from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.orm import ( + InstanceState, Mapper, ORMExecuteState, SessionTransaction, @@ -105,6 +118,25 @@ _installed: dict[type[Session], AuditTrail] = {} +class AsyncLoadError(MissingGreenlet): + """An audited attribute needed a load that ``AsyncSession`` cannot run. + + Raised in place of SQLAlchemy's ``MissingGreenlet`` (or, with psycopg, + the ``StatementError`` wrapping it) when host code touches, outside an + awaited call: + + - an expired or unloaded column of an ``Audited`` instance, read or + assigned (``Audited`` loads an expired column's old value when it is + assigned); + - a collection named in the model's ``track_relationships`` that is not + loaded. + + The message names the model and attribute and how to load it. The + original exception is the ``__cause__``. A subclass of + ``MissingGreenlet``, so existing handlers of that error still catch it. + """ + + class _CachedRow(NamedTuple): owner: SessionTransaction row: TransactionRow @@ -151,6 +183,7 @@ def install( listener = _Listener(trail) event.listen(cls, "after_flush", listener.after_flush) event.listen(cls, "do_orm_execute", listener.do_orm_execute) + event.listen(cls, "do_orm_execute", _explain_async_load) event.listen(cls, "after_transaction_end", _after_transaction_end) event.listen(cls, "after_soft_rollback", _after_soft_rollback) event.listen(cls, "after_rollback", _after_rollback) @@ -369,6 +402,110 @@ def _entry( } +# Private SQLAlchemy load option holding the instance of a column load (2.0 +# and 2.1); without it the message names only the model. +_REFRESH_STATE = "_refresh_state" + + +def _explain_async_load(orm_execute_state: ORMExecuteState) -> Result[Any] | None: + # Runs for every ORM statement of an installed session class, so the + # common case (not a lazy or expired-attribute load of an Audited + # instance on an async dialect) returns before running anything, and a + # session bound to a sync engine returns first. + state = orm_execute_state + bind = state.session.bind + if bind is not None and not bind.dialect.is_async: + return None + if state.is_column_load: + instance: InstanceState[Any] | None = getattr( + state.load_options, _REFRESH_STATE, None + ) + mapper = state.bind_mapper if instance is None else instance.mapper + elif state.is_relationship_load: + instance = state.lazy_loaded_from + if instance is None: + return None # an eager loader, run by an awaited query + mapper = instance.mapper + else: + return None + if mapper is None or not issubclass(mapper.class_, Audited): + return None + if bind is None: # bound per mapper or table + bind = state.session.get_bind(**state.bind_arguments) + if not bind.dialect.is_async: + return None + try: + return state.invoke_statement() + except (MissingGreenlet, StatementError) as exc: + cause = exc if isinstance(exc, MissingGreenlet) else exc.orig + if not isinstance(cause, MissingGreenlet) or isinstance(exc, AsyncLoadError): + raise + message = ( + _column_load_message(mapper, instance) + if state.is_column_load + else _collection_load_message(mapper, instance, state.bind_mapper) + ) + if message is None: + raise + raise AsyncLoadError(message) from exc + + +_NO_IMPLICIT_IO = ( + "loading it needs database I/O, which an AsyncSession cannot run outside " + "an awaited call" +) + + +def _column_load_message( + mapper: Mapper[Any], instance: InstanceState[Any] | None +) -> str: + name = mapper.class_.__qualname__ + columns = {prop.key for prop in mapper.column_attrs} + unloaded = [] if instance is None else sorted(instance.unloaded & columns) + if not unloaded: + what = f"an attribute of {name} is" + elif len(unloaded) == 1: + what = f"{name}.{unloaded[0]} is" + else: + what = ", ".join(f"{name}.{key}" for key in unloaded) + " are" + return ( + f"{what} not loaded (expired, e.g. by commit), and {_NO_IMPLICIT_IO}. " + "This happens when an unloaded attribute is read, and for an audited " + "column also when it is assigned: Audited loads the old value to " + "record the change. Load it first with `await session.refresh(obj)`, " + "or create the session with expire_on_commit=False." + ) + + +def _collection_load_message( + mapper: Mapper[Any], + instance: InstanceState[Any] | None, + target: Mapper[Any] | None, +) -> str | None: + # SQLAlchemy does not say which relationship is being loaded: name the + # tracked ones that are unloaded and point at the loaded class. + if instance is None or target is None: + return None + tracked = options_of(mapper.class_).track_relationships + keys = [ + rel.key + for rel in mapper.relationships + if rel.key in tracked + and rel.key in instance.unloaded + and (target.isa(rel.mapper) or rel.mapper.isa(target)) + ] + if not keys: + return None + name = mapper.class_.__qualname__ + attributes = " or ".join(f"{name}.{key}" for key in keys) + loaders = " or ".join(f"selectinload({name}.{key})" for key in keys) + return ( + f"{attributes} is not loaded, and {_NO_IMPLICIT_IO}. Load tracked " + f"collections eagerly before changing them: {loaders} in the query, " + 'or lazy="selectin" on the relationship.' + ) + + def _audited_model(state: ORMExecuteState) -> type[Any] | None: # Matched by table, which covers ORM update(Model) and Core update(table). table = getattr(state.statement, "table", None) diff --git a/docs/limitations.md b/docs/limitations.md index 3159383..4a72f72 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -32,9 +32,16 @@ Rows deleted by the database, through `ForeignKey(..., ondelete="CASCADE")` with - a column filled by the database on INSERT (`server_default`, `Computed`, `Identity`) in an `entity.created` entry, **only** when the mapper sets `eager_defaults=False`. SQLAlchemy's default, `eager_defaults="auto"`, fetches these values with `RETURNING` on PostgreSQL, so the entry holds the real value; - a column that is not loaded when the object is deleted, such as a `deferred()` column, in an `entity.deleted` entry. -## Tracked collections under AsyncSession +## Unloaded attributes under AsyncSession -Appending to or replacing a relationship collection that is not loaded makes SQLAlchemy load it first. `AsyncSession` cannot lazy-load, so the change fails with SQLAlchemy's `MissingGreenlet` error before the audit trail is involved. Load tracked collections eagerly (`selectinload(...)`, or `lazy="selectin"` on the relationship) before changing them. See the [async quickstart](quickstart.md#async). +`AsyncSession` cannot load an attribute implicitly: outside an awaited call, a load needs database I/O that SQLAlchemy cannot run, and it raises `MissingGreenlet`. Two such loads come from using the audit trail, and for them the library raises `AsyncLoadError` instead, naming the model, the attribute and the fix. The original error is its `__cause__`. + +- **A tracked collection that is not loaded.** Appending to or replacing a collection makes SQLAlchemy load it first. Load collections named in `track_relationships` eagerly before changing them: `selectinload(Post.tags)` in the query, or `lazy="selectin"` on the relationship. See the [async quickstart](quickstart.md#async). +- **An expired column of an `Audited` model.** To record the old value, `Audited` loads an expired column when it is assigned, not only when it is read. With the default `expire_on_commit=True`, every column is expired after `await session.commit()`, so `post.title = "new"` right after a commit needs a load. Refresh the object first (`await session.refresh(post)`), or create the session with `async_sessionmaker(..., expire_on_commit=False)`. + +`AsyncLoadError` subclasses SQLAlchemy's `MissingGreenlet`, so an `except MissingGreenlet` still catches it. With psycopg, SQLAlchemy wraps `MissingGreenlet` in a `StatementError`; for these two loads you get `AsyncLoadError` instead of that `StatementError`, on psycopg and asyncpg alike. Every other load (a relationship not in `track_relationships`, a model that is not `Audited`) keeps SQLAlchemy's own error. + +Inside the flush, the audit trail never loads anything: changes, labels, scopes and targets are built from what is already in memory. ## Relationships diff --git a/docs/quickstart.md b/docs/quickstart.md index 1900053..8d164ba 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -90,7 +90,7 @@ This example also tracks a relationship and points each task at its project as t ``` -`AsyncSession` cannot lazy-load, so load a tracked collection (here with `selectinload`) before you change it; see [limitations](limitations.md#tracked-collections-under-asyncsession). +`AsyncSession` cannot lazy-load, so load a tracked collection (here with `selectinload`) before you change it, and refresh an object expired by a commit before assigning to it; otherwise you get an `AsyncLoadError` naming the attribute. See [limitations](limitations.md#unloaded-attributes-under-asyncsession). Every query has an `a`-prefixed async variant: diff --git a/tests/db/test_async.py b/tests/db/test_async.py index 785a762..b31bbcf 100644 --- a/tests/db/test_async.py +++ b/tests/db/test_async.py @@ -2,8 +2,9 @@ The ``entity.*`` capture tests of ``test_listener*.py`` run under ``AsyncSession`` too (see ``listener_support``); these cover what only the -async API has: context variables set in coroutines, ``alog`` and the async -durable writer. ``async_engine`` runs each test on asyncpg and psycopg. +async API has: context variables set in coroutines, ``alog``, the async +durable writer and ``AsyncLoadError`` for loads outside the greenlet. +``async_engine`` runs each test on asyncpg and psycopg. """ from __future__ import annotations @@ -36,10 +37,12 @@ from sqlalchemy.orm import ( DeclarativeBase, Mapped, + ORMExecuteState, Session, mapped_column, relationship, selectinload, + sessionmaker, ) from audit_trail import ( @@ -50,9 +53,11 @@ AuditTrail, Severity, event, + listener, ) +from audit_trail.listener import AsyncLoadError from audit_trail.writer import AuditWriteError -from tests.db.listener_support import create_trail, logged_error +from tests.db.listener_support import create_trail, logged_error, record_statements CAP = 15 """Seconds after which a test that should not block fails instead of hanging.""" @@ -68,6 +73,7 @@ class AsyncEvent(AuditEvent): class Models: Tag: Any Post: Any + Comment: Any @pytest.fixture @@ -91,11 +97,17 @@ class Post(Base, Audited): id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(default="") tags: Mapped[list[Tag]] = relationship(secondary=post_tags) + comments: Mapped[list[Comment]] = relationship() # not tracked __audit__ = AuditOptions(track_relationships={"tags"}) + class Comment(Base): + __tablename__ = "comment" + id: Mapped[int] = mapped_column(primary_key=True) + post_id: Mapped[int] = mapped_column(ForeignKey("post.id")) + Base.metadata.create_all(engine) - return Models(Tag, Post) + return Models(Tag, Post, Comment) @dataclass @@ -318,26 +330,41 @@ async def test_alog_refuses_a_sync_durable_engine( assert env.activities() == [] -async def test_unloaded_tracked_collection_is_a_host_error( +def missing_greenlet(exc: BaseException | None) -> bool: + """``MissingGreenlet`` as raised by asyncpg, or wrapped by psycopg.""" + if isinstance(exc, StatementError) and not isinstance(exc, MissingGreenlet): + exc = exc.orig + return isinstance(exc, MissingGreenlet) and not isinstance(exc, AsyncLoadError) + + +async def seed(env: AsyncEnv, models: Models) -> None: + async with env.factory() as session: + session.add_all([models.Post(id=1, title="p"), models.Tag(id=1)]) + session.add(models.Comment(id=1, post_id=1)) + await session.commit() + + +async def test_unloaded_tracked_collection_is_explained( make: EnvMaker, models: Models ) -> None: # Appending to an unloaded collection loads it first, which is I/O - # outside the greenlet: SQLAlchemy raises MissingGreenlet in the host's - # code, before any flush. Hosts load tracked collections eagerly. + # outside the greenlet, in the host's code before any flush. The error + # names the tracked collection and how to load it. env = make() - async with env.factory() as session: - session.add_all([models.Post(id=1, title="p"), models.Tag(id=1)]) - await session.commit() + await seed(env, models) async with env.factory() as session: post = await session.get(models.Post, 1) tag = await session.get(models.Tag, 1) assert post is not None - # asyncpg raises it as is, psycopg wrapped in a StatementError. - with pytest.raises( - (MissingGreenlet, StatementError), match="greenlet_spawn has not been" - ): + with pytest.raises(AsyncLoadError) as raised: post.tags.append(tag) + message = str(raised.value) + assert "Post.tags is not loaded" in message + assert "selectinload(models..Post.tags)" in message + assert 'lazy="selectin"' in message + assert isinstance(raised.value, MissingGreenlet) + assert missing_greenlet(raised.value.__cause__) assert env.verbs() == ["entity.created"] async with env.factory() as session: @@ -349,3 +376,146 @@ async def test_unloaded_tracked_collection_is_a_host_error( await session.commit() updated = env.activities()[-1] assert updated["data"]["changes"] == {"tags": {"added": ["1"], "removed": []}} + + +async def test_expired_audited_column_is_explained( + make: EnvMaker, models: Models +) -> None: + # Audited keeps the old value of an assigned column (active_history), so + # assigning an expired column loads it: outside the greenlet that fails. + env = make() + await seed(env, models) + + async with env.factory() as session: + post = await session.get(models.Post, 1) + assert post is not None + await session.commit() # expires post + with pytest.raises(AsyncLoadError) as assigned: + post.title = "changed" + with pytest.raises(AsyncLoadError) as read: + _ = post.title + for raised in (assigned, read): + message = str(raised.value) + assert "Post.id, models..Post.title are not loaded" in message + assert "await session.refresh(obj)" in message + assert "expire_on_commit=False" in message + assert missing_greenlet(raised.value.__cause__) + + await session.refresh(post) + post.title = "changed" + await session.commit() + assert env.activities()[-1]["data"]["changes"] == {"title": ["p", "changed"]} + + +async def test_expired_column_without_the_instance_names_the_model( + make: EnvMaker, models: Models, monkeypatch: pytest.MonkeyPatch +) -> None: + # The instance of a column load comes from a private SQLAlchemy option; + # should it disappear, the error still is AsyncLoadError. + monkeypatch.setattr(listener, "_REFRESH_STATE", "_no_such_option") + env = make() + await seed(env, models) + async with env.factory() as session: + post = await session.get(models.Post, 1) + assert post is not None + await session.commit() + with pytest.raises(AsyncLoadError) as raised: + post.title = "changed" + assert "an attribute of models..Post is not loaded" in str(raised.value) + assert missing_greenlet(raised.value.__cause__) + + +async def test_loads_the_library_does_not_cause_keep_their_error( + make: EnvMaker, models: Models +) -> None: + env = make() + await seed(env, models) + + async with env.factory() as session: + post = await session.get(models.Post, 1) + tag = await session.get(models.Tag, 1) + assert post is not None + assert tag is not None + with pytest.raises((MissingGreenlet, StatementError)) as untracked: + _ = post.comments # a relationship not in track_relationships + await session.commit() # expires tag, which is not Audited + with pytest.raises((MissingGreenlet, StatementError)) as unaudited: + _ = tag.id + for raised in (untracked, unaudited): + assert missing_greenlet(raised.value), raised.value + + +async def test_audit_flush_emits_no_select( + make: EnvMaker, models: Models, async_engine: AsyncEngine +) -> None: + # Under AsyncSession the flush runs in the greenlet, where a lazy load + # would succeed silently: only the statements show that none happens. + env = make() + await seed(env, models) + async with env.factory() as session: + post = await session.get(models.Post, 1) + tag = await session.get(models.Tag, 1) + assert post is not None + await session.refresh(post, ["tags"]) + with record_statements(async_engine.sync_engine) as log: + log.recording = True + post.title = "changed" + post.tags.append(tag) + await session.commit() + assert log.statements + assert [s for s in log.statements if s.lstrip().upper().startswith("SELECT")] == [] + + +@pytest.mark.parametrize("per_mapper", [False, True], ids=["bind", "binds"]) +def test_sync_loads_skip_the_async_translation( + engine: Engine, + schema: str, + models: Models, + monkeypatch: pytest.MonkeyPatch, + per_mapper: bool, +) -> None: + trail = create_trail(engine, schema) + session_class: type[Session] = type("AuditedSyncSession", (Session,), {}) + factory = sessionmaker(class_=session_class) + trail.install(factory) + with factory(bind=engine) as session: + session.add_all([models.Post(id=1, title="p"), models.Tag(id=1)]) + session.commit() + + invoked: list[ORMExecuteState] = [] + invoke = ORMExecuteState.invoke_statement + + def spy(self: ORMExecuteState, *args: Any, **kwargs: Any) -> Any: + invoked.append(self) + return invoke(self, *args, **kwargs) + + monkeypatch.setattr(ORMExecuteState, "invoke_statement", spy) + if per_mapper: + session = factory(binds={models.Post: engine, models.Tag: engine}) + else: + session = factory(bind=engine) + with session: + post = session.get(models.Post, 1) + assert post is not None + session.expire(post) + assert post.tags == [] # tracked collection lazy load + post.title = "changed" # expired column load + assert len(invoked) == 0 + session.rollback() + + +async def test_async_session_bound_per_mapper_is_explained( + make: EnvMaker, models: Models, async_engine: AsyncEngine +) -> None: + env = make() + await seed(env, models) + factory = async_sessionmaker( + binds={models.Post: async_engine}, + sync_session_class=env.factory.kw["sync_session_class"], + ) + async with factory() as session: + post = await session.get(models.Post, 1) + assert post is not None + session.expire(post) + with pytest.raises(AsyncLoadError, match=r"Post\.title are not loaded"): + post.title = "changed" diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py index 608c3ac..c485ae2 100644 --- a/tests/unit/test_query.py +++ b/tests/unit/test_query.py @@ -100,12 +100,15 @@ def test_invalid_default_window(window: object, error: type[Exception]) -> None: def test_trail_passes_the_default_window() -> None: window = timedelta(days=30) trail = AuditTrail( - create_engine("postgresql+psycopg://"), default_query_window=window + create_engine("postgresql+psycopg://"), events=[], default_query_window=window ) assert trail.query.default_window == window assert ( - AuditTrail(create_engine("postgresql+psycopg://")).query.default_window is None + AuditTrail( + create_engine("postgresql+psycopg://"), events=[] + ).query.default_window + is None )