diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c2cd6c6..82a9368b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ All notable changes to this project are documented in this file. The format is b ## [Unreleased] ### Added +- Request-scoped database sessions now expose `session.on_commit(callback)` for + synchronous or asynchronous cache refreshes and other derived state. The + framework invokes callbacks only after a successful commit and discards them + on rollback or commit failure. - Every `smpy new` scaffold now ships Docker assets by default: a multi-stage `docker/host.Dockerfile` (uv + Node builder that runs `gen-pages` before the Vite build, slim non-root runtime that applies migrations on start), a diff --git a/docs/database/sessions.md b/docs/database/sessions.md index 457ee3af..2cea3ea2 100644 --- a/docs/database/sessions.md +++ b/docs/database/sessions.md @@ -25,10 +25,10 @@ Usage in endpoints: ```python from typing import Annotated from fastapi import Depends -from sqlalchemy.ext.asyncio import AsyncSession +from simple_module_db import RequestSession from simple_module_db.deps import get_db -SessionDep = Annotated[AsyncSession, Depends(get_db)] +SessionDep = Annotated[RequestSession, Depends(get_db)] @router.post("") @@ -88,6 +88,33 @@ async def create(self, data: OrderCreate) -> OrderOut: Flushing sends the INSERT but keeps the transaction open. Rollback still works until the dependency commits. +## Refreshing derived state after commit + +The request session exposes `on_commit()` for cache invalidation and other derived +in-process state that must never reflect an uncommitted transaction: + +```python +async def create(self, data: OrderCreate) -> OrderOut: + order = Order(**data.model_dump()) + self.session.add(order) + await self.session.flush() + self.session.on_commit(self.order_cache.refresh) + return OrderOut.model_validate(order) +``` + +Callbacks take no arguments and may be synchronous or asynchronous. The framework +runs them in registration order after a successful commit and before the response +leaves the server. It discards them when the request rolls back, when commit fails, +or when the request has no writes. A callback should refresh derived state, not +perform more database mutations. + +At callback time the original transaction is already durable, so callback errors +cannot roll it back. The framework logs a `db.session.on_commit_failed` error, +continues with remaining callbacks, and preserves the successful response. + +`on_commit()` belongs to the framework-managed request lifecycle. For a manually +managed session, perform post-commit work explicitly after the transaction block. + ## Manual transactions If you need finer control — e.g. a background worker that processes many items in its own transactions — use the `DatabaseState.session_factory` directly: diff --git a/docs/framework-conventions.md b/docs/framework-conventions.md index 9cf27e9b..2d7db48a 100644 --- a/docs/framework-conventions.md +++ b/docs/framework-conventions.md @@ -193,7 +193,10 @@ Each request opens one session: - Read-only requests exit via rollback — cheaper, and keeps the session out of write-side profiling. - Exceptions always rollback. -Service code should not call `session.commit()` directly. Flush for intermediate reads if you need DB-assigned values, then let the framework commit. +Service code should not call `session.commit()` directly. Flush for intermediate reads if you need DB-assigned values, then let the framework commit. The injected +`RequestSession` also provides `session.on_commit(callback)` for synchronous or +asynchronous cache refreshes that must only observe durable state. Callbacks run +after successful commit and are discarded on rollback or commit failure. The commit lands in `CommitBeforeResponseMiddleware`, which intercepts the ASGI `http.response.start` message — the last point still inside the request. That is diff --git a/framework/db/README.md b/framework/db/README.md index c1ea174d..674a4a76 100644 --- a/framework/db/README.md +++ b/framework/db/README.md @@ -11,7 +11,8 @@ pip install simple_module_db ## What it provides - `create_module_base("")` — a module-scoped declarative `Base` with its own `MetaData`. All modules share the host's single schema, so `__tablename__` should be prefixed with the module name (e.g. `users_user`) to avoid collisions. -- Per-request async session (`get_db`) with an auto-commit-on-flush hook — `after_flush` commits if there are pending writes, rolls back otherwise. +- Per-request `RequestSession` (`get_db`) that commits pending writes before the response and rolls back read-only requests. +- `session.on_commit(callback)` for synchronous or asynchronous cache refreshes that must only run after a successful request commit. - Mixins in `simple_module_db.mixins`: `AuditMixin` (created_at/updated_at), `SoftDeleteMixin` (auto-filtered unless `stmt.execution_options(include_deleted=True)`), `MultiTenantMixin`, `VersionedMixin`. - `DatabaseState` container used by the framework to avoid global mutable state. @@ -34,12 +35,13 @@ class Order(Base, AuditMixin, SoftDeleteMixin, table=True): In a service: ```python -from simple_module_db import get_db +from simple_module_db import RequestSession, get_db -async def create_order(session = Depends(get_db), ...): +async def create_order(session: RequestSession = Depends(get_db), ...): order = Order(customer_id=..., total_cents=...) session.add(order) - await session.flush() # assigns order.id; auto-commit happens after the request + await session.flush() # assigns order.id; commit happens before the response + session.on_commit(order_cache.refresh) return order ``` diff --git a/framework/db/simple_module_db/__init__.py b/framework/db/simple_module_db/__init__.py index 5da43338..7a91b601 100644 --- a/framework/db/simple_module_db/__init__.py +++ b/framework/db/simple_module_db/__init__.py @@ -2,6 +2,7 @@ from simple_module_db.audit import AuditRecord from simple_module_db.base import create_module_base +from simple_module_db.callbacks import OnCommitCallback from simple_module_db.deps import get_db from simple_module_db.listeners import TenantIsolationError, current_tenant_id from simple_module_db.migrations import ( @@ -13,7 +14,7 @@ from simple_module_db.mixins import AuditMixin, MultiTenantMixin, SoftDeleteMixin, VersionedMixin from simple_module_db.provider import DatabaseProvider, detect_provider from simple_module_db.search import LIKE_ESCAPE_CHAR, like_contains_pattern, like_prefix_pattern -from simple_module_db.session import DatabaseState, init_db +from simple_module_db.session import DatabaseState, RequestSession, init_db from simple_module_db.transaction import CommitBeforeResponseMiddleware, finalize_session __all__ = [ @@ -24,6 +25,8 @@ "DatabaseProvider", "DatabaseState", "MultiTenantMixin", + "OnCommitCallback", + "RequestSession", "SoftDeleteMixin", "TenantIsolationError", "VersionedMixin", diff --git a/framework/db/simple_module_db/callbacks.py b/framework/db/simple_module_db/callbacks.py new file mode 100644 index 00000000..4c7024cd --- /dev/null +++ b/framework/db/simple_module_db/callbacks.py @@ -0,0 +1,47 @@ +"""Post-commit callbacks for framework-managed database sessions.""" + +from __future__ import annotations + +import inspect +import logging +from collections.abc import Awaitable, Callable + +from sqlalchemy.ext.asyncio import AsyncSession + +type OnCommitCallback = Callable[[], Awaitable[None] | None] + +_CALLBACKS_KEY = "sm_db_on_commit_callbacks" +logger = logging.getLogger("simple_module.db") + + +def register_on_commit(session: AsyncSession, callback: OnCommitCallback) -> None: + """Queue ``callback`` for this session's next successful finalization.""" + if not callable(callback): + raise TypeError("on_commit callback must be callable") + callbacks = session.info.setdefault(_CALLBACKS_KEY, []) + callbacks.append(callback) + + +def discard_on_commit_callbacks(session: AsyncSession) -> None: + """Discard callbacks belonging to a transaction that did not commit.""" + session.info.pop(_CALLBACKS_KEY, None) + + +async def run_on_commit_callbacks(session: AsyncSession) -> None: + """Run and remove queued callbacks after a successful commit. + + The transaction is already durable, so callback failures are logged and do + 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: + try: + result = callback() + if inspect.isawaitable(result): + await result + except Exception: + logger.exception( + "db.session.on_commit_failed", + extra={"operation": "on_commit_failed"}, + ) diff --git a/framework/db/simple_module_db/deps.py b/framework/db/simple_module_db/deps.py index 47dd0f39..4fbf39ff 100644 --- a/framework/db/simple_module_db/deps.py +++ b/framework/db/simple_module_db/deps.py @@ -6,8 +6,8 @@ from collections.abc import AsyncGenerator from fastapi import Request -from sqlalchemy.ext.asyncio import AsyncSession +from simple_module_db.session import RequestSession from simple_module_db.transaction import ( SESSION_START_KEY, finalize_session, @@ -16,7 +16,7 @@ ) -async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]: +async def get_db(request: Request) -> AsyncGenerator[RequestSession, None]: """Yield an async database session, auto-closing on exit. Commits only when the session has pending writes (``new``, ``dirty``, @@ -34,7 +34,7 @@ async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]: Usage in FastAPI endpoints:: @router.get("/items") - async def list_items(db: AsyncSession = Depends(get_db)): + async def list_items(db: RequestSession = Depends(get_db)): ... """ factory = request.app.state.sm.db.session_factory diff --git a/framework/db/simple_module_db/session.py b/framework/db/simple_module_db/session.py index d32c0ae9..4eac7a5a 100644 --- a/framework/db/simple_module_db/session.py +++ b/framework/db/simple_module_db/session.py @@ -13,15 +13,28 @@ ) from sqlalchemy.orm import Session +from simple_module_db.callbacks import OnCommitCallback, register_on_commit from simple_module_db.provider import DatabaseProvider, detect_provider +class RequestSession(AsyncSession): + """Async session with hooks for the request-owned transaction.""" + + def on_commit(self, callback: OnCommitCallback) -> None: + """Run ``callback`` after this request's next successful commit. + + Callbacks may be synchronous or asynchronous. They are discarded when + the request rolls back or has no writes. + """ + register_on_commit(self, callback) + + @dataclass class DatabaseState: """Holds all database state for a single application instance.""" engine: AsyncEngine - session_factory: async_sessionmaker[AsyncSession] + session_factory: async_sessionmaker[RequestSession] sync_session_class: type[Session] = field(repr=False, default=Session) audit_callback: Callable | None = field(default=None, repr=False) _listeners_registered: bool = field(default=False, repr=False) @@ -67,7 +80,10 @@ def init_db( # Scoped Session subclass so event listeners only fire for this engine's sessions scoped_session_class = type("ScopedSession", (Session,), {}) session_factory = async_sessionmaker( - engine, expire_on_commit=False, sync_session_class=scoped_session_class + engine, + class_=RequestSession, + expire_on_commit=False, + sync_session_class=scoped_session_class, ) return DatabaseState( diff --git a/framework/db/simple_module_db/transaction.py b/framework/db/simple_module_db/transaction.py index 8871806b..5e3d012a 100644 --- a/framework/db/simple_module_db/transaction.py +++ b/framework/db/simple_module_db/transaction.py @@ -31,6 +31,10 @@ from sqlalchemy.ext.asyncio import AsyncSession +from simple_module_db.callbacks import ( + discard_on_commit_callbacks, + run_on_commit_callbacks, +) from simple_module_db.listeners import SESSION_HAS_WRITES_KEY logger = logging.getLogger("simple_module.db") @@ -97,6 +101,7 @@ async def finalize_session(session: AsyncSession) -> None: if session.info.get(_FINALIZED_KEY): return _settle(session) + discard_on_commit_callbacks(session) await session.rollback() logger.debug( "db.session.read_only", @@ -106,6 +111,7 @@ async def finalize_session(session: AsyncSession) -> None: try: await session.commit() except Exception: + discard_on_commit_callbacks(session) await session.rollback() raise finally: @@ -116,10 +122,12 @@ async def finalize_session(session: AsyncSession) -> None: "db.session.commit", extra={"operation": "commit", "db_duration_ms": _elapsed_ms(session)}, ) + await run_on_commit_callbacks(session) async def rollback_session(session: AsyncSession) -> None: """Roll ``session`` back and drop its pending work. Idempotent.""" + discard_on_commit_callbacks(session) if session.info.get(_FINALIZED_KEY) and not _has_pending(session): return _settle(session) diff --git a/framework/db/tests/test_session.py b/framework/db/tests/test_session.py index 9c7bb229..6f0adb03 100644 --- a/framework/db/tests/test_session.py +++ b/framework/db/tests/test_session.py @@ -6,7 +6,7 @@ from unittest.mock import MagicMock from simple_module_db.deps import get_db -from simple_module_db.session import DatabaseState, init_db +from simple_module_db.session import DatabaseState, RequestSession, init_db from sqlalchemy.ext.asyncio import AsyncSession @@ -58,6 +58,7 @@ async def test_get_db_yields_session(self): gen = get_db(mock_request) session = await gen.__anext__() assert isinstance(session, AsyncSession) + assert isinstance(session, RequestSession) with contextlib.suppress(StopAsyncIteration): await gen.__anext__() diff --git a/framework/db/tests/test_transaction.py b/framework/db/tests/test_transaction.py index 574a1db8..9edb3953 100644 --- a/framework/db/tests/test_transaction.py +++ b/framework/db/tests/test_transaction.py @@ -20,6 +20,7 @@ from _models import _TxnBase, _TxnThing from fastapi import BackgroundTasks, Depends, FastAPI from fastapi.responses import StreamingResponse +from simple_module_db import OnCommitCallback, RequestSession from simple_module_db.deps import get_db from simple_module_db.listeners import register_listeners from simple_module_db.session import init_db @@ -28,7 +29,12 @@ from sqlmodel import select -def _build_app(db_state, *, with_middleware: bool = True) -> FastAPI: +def _build_app( + db_state, + *, + with_middleware: bool = True, + on_commit: OnCommitCallback | None = None, +) -> FastAPI: """A miniature host: create flushes only, read opens its own session.""" app = FastAPI() if with_middleware: @@ -36,12 +42,14 @@ def _build_app(db_state, *, with_middleware: bool = True) -> FastAPI: app.state.sm = SimpleNamespace(db=db_state) @app.post("/things", status_code=201) - async def create(name: str, db: AsyncSession = Depends(get_db)): + async def create(name: str, db: RequestSession = Depends(get_db)): # Deliberately no commit() — get_db owns the unit of work, which is # exactly the pattern the framework documents for service code. thing = _TxnThing(name=name) db.add(thing) await db.flush() + if on_commit is not None: + db.on_commit(on_commit) return {"id": thing.id} @app.get("/things/{thing_id}") @@ -54,9 +62,11 @@ async def read(thing_id: int, db: AsyncSession = Depends(get_db)): return {"found": True, "name": found.name} @app.post("/boom", status_code=201) - async def boom(db: AsyncSession = Depends(get_db)): + async def boom(db: RequestSession = Depends(get_db)): db.add(_TxnThing(name="doomed")) await db.flush() + if on_commit is not None: + db.on_commit(on_commit) raise RuntimeError("endpoint blew up after writing") return app @@ -131,11 +141,54 @@ async def spy(message): "row was not committed by the time the response left the server" ) - async def test_endpoint_exception_still_rolls_back(self, db_state): - """The middleware must not turn a failed request's writes into a commit.""" - async with await _client(_build_app(db_state)) as client: + async def test_on_commit_runs_after_the_write_is_durable(self, db_state): + """A cache refresh can only observe state committed in another session.""" + observed: list[list[str]] = [] + + async def refresh_cache(): + async with db_state.session_factory() as other: + rows = (await other.execute(select(_TxnThing))).scalars().all() + observed.append([row.name for row in rows]) + + async with await _client(_build_app(db_state, on_commit=refresh_cache)) as client: + assert (await client.post("/things", params={"name": "committed"})).status_code == 201 + + assert observed == [["committed"]] + + async def test_on_commit_failure_does_not_skip_later_callbacks(self, db_state): + """The commit is durable, so callback errors are logged rather than returned as 500s.""" + called: list[bool] = [] + app = _build_app(db_state) + + @app.post("/callback-errors", status_code=201) + async def callback_errors(db: RequestSession = Depends(get_db)): + db.add(_TxnThing(name="durable-despite-cache-error")) + await db.flush() + + def fail_refresh(): + raise RuntimeError("cache unavailable") + + db.on_commit(fail_refresh) + db.on_commit(lambda: called.append(True)) + return {"ok": True} + + async with await _client(app) as client: + response = await client.post("/callback-errors") + + assert response.status_code == 201 + assert called == [True] + async with db_state.session_factory() as session: + names = [row.name for row in (await session.execute(select(_TxnThing))).scalars()] + assert names == ["durable-despite-cache-error"] + + async def test_endpoint_exception_rolls_back_without_on_commit(self, db_state): + """Callbacks for work discarded by an endpoint failure never run.""" + called: list[bool] = [] + app = _build_app(db_state, on_commit=lambda: called.append(True)) + async with await _client(app) as client: assert (await client.post("/boom")).status_code == 500 + assert called == [] async with db_state.session_factory() as session: assert (await session.execute(select(_TxnThing))).scalars().all() == [] @@ -154,11 +207,14 @@ async def explode(self, *args, **kwargs): monkeypatch.setattr(AsyncSession, "commit", explode) - async with await _client(_build_app(db_state)) as client: + called: list[bool] = [] + app = _build_app(db_state, on_commit=lambda: called.append(True)) + async with await _client(app) as client: response = await client.post("/things", params={"name": "nope"}) assert response.status_code == 500 assert response.json() == {"detail": "Internal Server Error"} + assert called == [] monkeypatch.undo() async with db_state.session_factory() as session: @@ -216,10 +272,16 @@ async def body(): async def test_still_commits_without_the_middleware(self, db_state): """get_db keeps its own fallback finalize, so the dependency works standalone — in a WebSocket handler, or a test that builds no stack.""" - app = _build_app(db_state, with_middleware=False) + called: list[bool] = [] + app = _build_app( + db_state, + with_middleware=False, + on_commit=lambda: called.append(True), + ) async with await _client(app) as client: assert (await client.post("/things", params={"name": "solo"})).status_code == 201 + assert called == [True] async with db_state.session_factory() as session: names = [t.name for t in (await session.execute(select(_TxnThing))).scalars().all()] assert names == ["solo"]