Tech-debt batch 11: framework/db — connection-bound policy, savepoint on_commit, soft-delete audit (#630 #631 #632)

CHANGELOG.md

@@ -14,6 +14,10 @@ All notable changes to this project are documented in this file. The format is b
 
 ## [Unreleased]
 
+### Fixed
+- Database: discard `on_commit` callbacks registered inside a rolled-back savepoint,
+  including released child savepoints, while preserving callbacks from surviving work (#631).
+
 ### Added
 - `tenants.contracts.UsageMeter`: an implementation-free usage-metering protocol next to
   `EntitlementProvider` (`check`, `consume`, `record`, `refund`, `reserve`/`commit`/`release`,
@@ -23,6 +27,8 @@ All notable changes to this project are documented in this file. The format is b
   session (#615).
 
 ### Security
+- DB: sessions bound to a SQLAlchemy `Connection` now inherit its engine's tenant
+  policy and audit configuration, preserving strict tenant isolation and audit callbacks (#630).
 - Sign-in now rotates the session: users' `on_after_login` (password, OAuth, accept-invite)
   clears it and keeps only the pending post-login redirect, so a CSRF token (or any other
   session state) issued before sign-in no longer survives it. Keycloak sign-in rotates the CSRF
@@ -72,6 +78,8 @@ All notable changes to this project are documented in this file. The format is b
   tracked for removal in 0.1.0 (#594).
 
 ### Fixed
+- DB audit: persistent rows deleted through the soft-delete listener now emit one
+  `soft_deleted` record, including when business fields were edited before deletion (#632).
 - Users: read-only demo mode now follows the resolved identity, not only the session cookie. Writes made with a bearer access token or personal API token of a demo account are refused (403 `DEMO_READ_ONLY`, or the Inertia 409 redirect) through the new `AuthState.principal_guards` hook, and demo accounts can no longer mint tokens via `POST /api/users/auth/token` or `/token/refresh` while `demo_read_only` is on (#611).
 - Users: the default user role is created at startup when missing, so an app scaffolded with
   `smpy new` (schema baseline, no data migrations) can register users (#600).

framework/db/simple_module_db/audit.py

@@ -116,6 +116,23 @@ def _serialize(value: Any) -> Any:
     return str(value)
 
 
+def _column_diffs(
+    obj: object, inspector: Any, excl: frozenset[str], pk_cols: set[str]
+) -> list[dict[str, Any]]:
+    """Old/new diffs for the changed, non-excluded, non-PK columns of *obj*."""
+    changes: list[dict[str, Any]] = []
+    for col_name in _column_names(obj):
+        if col_name in excl or col_name in pk_cols:
+            continue
+        hist = inspector.attrs[col_name].history
+        if not hist.has_changes():
+            continue
+        old_val = hist.deleted[0] if hist.deleted else None
+        new_val = hist.added[0] if hist.added else None
+        changes.append({"field": col_name, "old": _serialize(old_val), "new": _serialize(new_val)})
+    return changes
+
+
 # ---------------------------------------------------------------------------
 # Two-phase capture
 # ---------------------------------------------------------------------------
@@ -155,21 +172,6 @@ def _get_pk_cols(obj: object) -> set[str]:
         if _is_excluded(obj):
             continue
 
-        # Soft-delete listener moves objects from session.deleted → session.new
-        # with is_deleted=True.  Classify them as "soft_deleted", not "created".
-        if isinstance(obj, SoftDeleteMixin) and getattr(obj, "is_deleted", False):
-            pending.append(
-                _PendingChange(
-                    obj_ref=obj,
-                    entity_type=type(obj).__name__,
-                    action="soft_deleted",
-                    changes=[],
-                    user_id=user_id,
-                    correlation_id=correlation_id,
-                )
-            )
-            continue
-
         excl = _get_excluded(obj)
         pk_cols = _get_pk_cols(obj)
         changes: list[dict[str, Any]] = []
@@ -194,21 +196,26 @@ def _get_pk_cols(obj: object) -> set[str]:
             continue
         if _is_excluded(obj):
             continue
-        excl = _get_excluded(obj)
-        pk_cols = _get_pk_cols(obj)
         inspector = sa_inspect(obj)
-        changes = []
-        for col_name in _column_names(obj):
-            if col_name in excl or col_name in pk_cols:
-                continue
-            hist = inspector.attrs[col_name].history
-            if not hist.has_changes():
-                continue
-            old_val = hist.deleted[0] if hist.deleted else None
-            new_val = hist.added[0] if hist.added else None
-            changes.append(
-                {"field": col_name, "old": _serialize(old_val), "new": _serialize(new_val)}
+        changes = _column_diffs(obj, inspector, _get_excluded(obj), _get_pk_cols(obj))
+        # The soft-delete listener re-adds persistent objects as dirty. Edits made in
+        # the same flush stay on the soft_deleted record.
+        if (
+            isinstance(obj, SoftDeleteMixin)
+            and obj.is_deleted
+            and inspector.attrs.is_deleted.history.has_changes()
+        ):
+            pending.append(
+                _PendingChange(
+                    obj_ref=obj,
+                    entity_type=type(obj).__name__,
+                    action="soft_deleted",
+                    changes=changes,
+                    user_id=user_id,
+                    correlation_id=correlation_id,
+                )
             )
+            continue
         if changes:
             pending.append(
                 _PendingChange(

framework/db/simple_module_db/callbacks.py

@@ -6,9 +6,12 @@
 import logging
 from collections.abc import Awaitable, Callable
 
+from sqlalchemy import event
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Session, SessionTransaction
 
 type OnCommitCallback = Callable[[], Awaitable[None] | None]
+type _CallbackEntry = tuple[set[SessionTransaction], OnCommitCallback]
 
 _CALLBACKS_KEY = "sm_db_on_commit_callbacks"
 logger = logging.getLogger("simple_module.db")
@@ -18,8 +21,27 @@ def register_on_commit(session: AsyncSession, callback: OnCommitCallback) -> Non
     """Queue ``callback`` for this session's next successful finalization."""
     if not callable(callback):
         raise TypeError("on_commit callback must be callable")
+    sync_session = session.sync_session
+    if not event.contains(sync_session, "after_soft_rollback", _discard_rolled_back_callbacks):
+        event.listen(sync_session, "after_soft_rollback", _discard_rolled_back_callbacks)
+    savepoints: set[SessionTransaction] = set()
+    transaction = sync_session.get_nested_transaction()
+    # Keep ancestors too: a released child still belongs to its parent's rollback.
+    while transaction is not None:
+        if transaction.nested:
+            savepoints.add(transaction)
+        transaction = transaction.parent
     callbacks = session.info.setdefault(_CALLBACKS_KEY, [])
-    callbacks.append(callback)
+    callbacks.append((savepoints, callback))
+
+
+def _discard_rolled_back_callbacks(session: Session, transaction: SessionTransaction) -> None:
+    if transaction.parent is None:
+        session.info.pop(_CALLBACKS_KEY, None)
+    elif transaction.nested and _CALLBACKS_KEY in session.info:
+        session.info[_CALLBACKS_KEY] = [
+            entry for entry in session.info[_CALLBACKS_KEY] if transaction not in entry[0]
+        ]
 
 
 def discard_on_commit_callbacks(session: AsyncSession) -> None:
@@ -34,8 +56,8 @@ async def run_on_commit_callbacks(session: AsyncSession) -> None:
     not turn a successful database mutation into a misleading failed response.
     Remaining callbacks still run.
     """
-    callbacks: list[OnCommitCallback] = session.info.pop(_CALLBACKS_KEY, [])
-    for callback in callbacks:
+    callbacks: list[_CallbackEntry] = session.info.pop(_CALLBACKS_KEY, [])
+    for _, callback in callbacks:
         try:
             result = callback()
             if inspect.isawaitable(result):

framework/db/simple_module_db/query_filter.py

@@ -16,7 +16,7 @@
 from typing import Any
 from weakref import WeakKeyDictionary
 
-from sqlalchemy.engine import Engine
+from sqlalchemy.engine import Connection, Engine
 from sqlalchemy.orm import ORMExecuteState, Session, with_loader_criteria
 
 from simple_module_db import model_registry as registry
@@ -73,6 +73,8 @@ def bind_engine_policy(engine: Engine, policy: Any) -> None:
 def engine_policy(session: Session) -> Any:
     """The policy bound to this session's engine, or ``None``."""
     bind = session.bind
+    if isinstance(bind, Connection):
+        bind = bind.engine
     return _engine_policy.get(bind) if isinstance(bind, Engine) else None
 
 

framework/db/tests/test_audit.py

@@ -18,21 +18,28 @@
     PartialExcludeModel,
     SoftDeleteItem,
 )
+from simple_module_core.correlation import correlation_id
 from simple_module_db.audit import (
     AuditRecord,
     collect_audit_records,
     finalize_records,
     snapshot_changes,
 )
-from simple_module_db.listeners import register_listeners
+from simple_module_db.listeners import current_user_id, register_listeners
 from simple_module_db.session import init_db
 from sqlalchemy.ext.asyncio import AsyncSession
 
 
 @pytest.fixture
-async def audit_session() -> AsyncGenerator[AsyncSession, None]:
+def audit_records() -> list[AuditRecord]:
+    return []
+
+
+@pytest.fixture
+async def audit_session(audit_records: list[AuditRecord]) -> AsyncGenerator[AsyncSession, None]:
     """Session backed by in-memory SQLite with listeners registered."""
     db_state = init_db("sqlite+aiosqlite:///:memory:")
+    db_state.audit_callback = lambda session, records: audit_records.extend(records)
     try:
         register_listeners(db_state)
         async with db_state.engine.begin() as conn:
@@ -191,48 +198,48 @@ def _collect(session):
 # ── collect_audit_records: soft-deleted entity ────────────────────────────
 
 
+@pytest.mark.parametrize("change_title", [False, True])
 async def test_soft_deleted_entity_produces_soft_deleted_record(
     audit_session: AsyncSession,
+    audit_records: list[AuditRecord],
+    change_title: bool,
 ):
-    """A SoftDeleteMixin object re-added to session.new with is_deleted=True
-    should produce action='soft_deleted', not action='created'."""
+    """Deleting a persistent row emits one soft_deleted record, even with edits."""
     item = SoftDeleteItem(title="doomed")
     audit_session.add(item)
     await audit_session.commit()
     await audit_session.refresh(item)
 
-    # Simulate what the soft-delete listener does: expunge from deleted,
-    # set is_deleted=True, re-add to session.new.  We use make_transient
-    # so SQLAlchemy treats the re-added object as new (matching the
-    # session state that triggers Bug 1).
-    await audit_session.delete(item)
-
-    def _simulate_soft_delete(session):
-        from sqlalchemy.orm import make_transient
-
-        session.expunge(item)
-        item.is_deleted = True
-        make_transient(item)
-        session.add(item)
-
-    await audit_session.run_sync(_simulate_soft_delete)
-
-    records: list[AuditRecord] = []
-
-    def _collect(session):
-        records.extend(collect_audit_records(session, user_id="admin", correlation_id="req-sd"))
-
-    await audit_session.run_sync(_collect)
+    audit_records.clear()
+    if change_title:
+        item.title = "edited before deletion"
+    user_token = current_user_id.set("admin")
+    correlation_token = correlation_id.set("req-sd")
+    try:
+        await audit_session.delete(item)
+        await audit_session.flush()
+    finally:
+        current_user_id.reset(user_token)
+        correlation_id.reset(correlation_token)
 
-    assert len(records) == 1
-    rec = records[0]
+    assert item.is_deleted is True
+    [rec] = audit_records
     assert rec.action == "soft_deleted"
     assert rec.entity_type == "SoftDeleteItem"
     assert rec.entity_id == str(item.id)
-    assert rec.changes == []
+    if change_title:
+        assert rec.changes == [{"field": "title", "old": "doomed", "new": "edited before deletion"}]
+    else:
+        assert rec.changes == []
     assert rec.user_id == "admin"
     assert rec.correlation_id == "req-sd"
 
+    audit_records.clear()
+    item.title = "edited after deletion"
+    await audit_session.flush()
+    [rec] = audit_records
+    assert rec.action == "updated"
+
 
 # ── collect_audit_records: soft-delete fields excluded from diffs ─────────
 

framework/db/tests/test_db_connection_policy.py

@@ -0,0 +1,67 @@
+"""Connection-bound sessions retain their engine's tenancy and audit policy (#630)."""
+
+from collections.abc import Iterator
+
+import pytest
+from _audit_models import AuditTestItem
+from _models import _TenantItem
+from simple_module_db.audit import AuditRecord
+from simple_module_db.listeners import attach_session_listeners
+from simple_module_db.query_filter import EngineTenancy, bind_engine_policy
+from simple_module_db.tenancy import TenantIsolationError
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+
+@pytest.fixture
+def connection_session() -> Iterator[tuple[Session, list[AuditRecord]]]:
+    engine = create_engine("sqlite://")
+    records: list[AuditRecord] = []
+
+    def capture(session: Session, changes: list[AuditRecord]) -> None:
+        records.extend(changes)
+
+    class ConnectionSession(Session):
+        pass
+
+    attach_session_listeners(ConnectionSession)
+    bind_engine_policy(engine, EngineTenancy(tenant_strict=True, audit_callback=capture))
+    try:
+        _TenantItem.__table__.create(engine)
+        AuditTestItem.__table__.create(engine)
+        with engine.begin() as connection:
+            connection.execute(
+                _TenantItem.__table__.insert(),
+                [{"name": "row-a", "tenant_id": "a"}, {"name": "row-b", "tenant_id": "b"}],
+            )
+        with engine.connect() as connection, ConnectionSession(bind=connection) as session:
+            yield session, records
+    finally:
+        engine.dispose()
+
+
+def test_connection_bound_select_without_tenant_fails_closed(connection_session):
+    session, _records = connection_session
+    with pytest.raises(TenantIsolationError):
+        session.execute(select(_TenantItem)).all()
+
+
+def test_connection_bound_flush_without_tenant_fails_closed(connection_session):
+    session, _records = connection_session
+    session.add(_TenantItem(name="missing-tenant"))
+    with pytest.raises(TenantIsolationError):
+        session.flush()
+
+
+def test_connection_bound_flush_delivers_audit_records(connection_session):
+    session, records = connection_session
+    item = AuditTestItem(name="audited")
+    session.add(item)
+    session.flush()
+
+    [record] = records
+    assert (record.entity_type, record.action, record.entity_id) == (
+        "AuditTestItem",
+        "created",
+        str(item.id),
+    )