feat: serve GET and HEAD from an autocommit, read-only engine (#84)

Closes #83

Author: lesnik512

AGENTS.md

@@ -40,8 +40,8 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 
 `app/ioc.py` defines providers:
 - `database_engine` — singleton-ish primary `AsyncEngine` (`create_primary_sa_engine`, `close_sa_engine` finalizer).
-- `database_replica_engine` — singleton-ish replica `AsyncEngine` built from `DB_REPLICA_DSN`, or `None` when it is unset.
-- `dynamic_engine` — `Scope.REQUEST`; `choose_sa_engine` returns the replica for `GET`/`HEAD` requests when one is configured, the primary otherwise (including when there is no request). GET handlers must not write, and a read right after a write may see replica lag.
+- `database_read_engine` — singleton-ish read `AsyncEngine` built from `DB_REPLICA_DSN`, or from `DB_DSN` when it is unset. It runs in `AUTOCOMMIT` (no BEGIN/ROLLBACK per request, and the pre-ping is a bare `;`) and sets `default_transaction_read_only=on`, so Postgres rejects writes on it. Each statement sees its own snapshot, so a multi-query read (e.g. `fetch_with_cards`) can see a write that commits in between. Without a replica it is a second pool on the primary, so a process can hold up to `2 × db_pool_size` connections.
+- `dynamic_engine` — `Scope.REQUEST`; `choose_sa_engine` returns the read engine for `GET`/`HEAD` requests, the primary otherwise (including when there is no request). GET handlers cannot write, and a read right after a write may see replica lag.
 - `session` — `Scope.REQUEST`, bound to `dynamic_engine`, finalized by `close_session`.
 - `decks_repository`, `cards_repository` — `Scope.REQUEST`, depend on `session`, configured with `auto_commit=True` (commit happens at session close, not per call).
 

app/ioc.py

@@ -6,7 +6,7 @@
     close_sa_engine,
     close_session,
     create_primary_sa_engine,
-    create_replica_sa_engine,
+    create_read_sa_engine,
     create_session,
 )
 
@@ -17,15 +17,15 @@ class Dependencies(Group):
         cache=providers.CacheSettings(finalizer=close_sa_engine),
         bound_type=None,
     )
-    database_replica_engine = providers.Factory(
-        creator=create_replica_sa_engine,
+    database_read_engine = providers.Factory(
+        creator=create_read_sa_engine,
         cache=providers.CacheSettings(finalizer=close_sa_engine),
         bound_type=None,
     )
     dynamic_engine = providers.Factory(
         scope=Scope.REQUEST,
         creator=choose_sa_engine,
-        kwargs={"primary_engine": database_engine, "replica_engine": database_replica_engine},
+        kwargs={"primary_engine": database_engine, "read_engine": database_read_engine},
     )
     session = providers.Factory(
         scope=Scope.REQUEST,

app/resources/db.py

@@ -12,41 +12,45 @@
 logger = logging.getLogger(__name__)
 
 
-REPLICA_METHODS: typing.Final = frozenset({"GET", "HEAD"})
+READ_METHODS: typing.Final = frozenset({"GET", "HEAD"})
 
 
-def create_sa_engine(url: URL) -> sa.AsyncEngine:
+def create_sa_engine(url: URL, *, read_only: bool = False) -> sa.AsyncEngine:
     return sa.create_async_engine(
         url=url,
         echo=settings.service_debug,
         echo_pool=settings.service_debug,
         pool_size=settings.db_pool_size,
         pool_pre_ping=settings.db_pool_pre_ping,
         max_overflow=settings.db_max_overflow,
+        isolation_level="AUTOCOMMIT" if read_only else None,
+        connect_args={"server_settings": {"default_transaction_read_only": "on"}} if read_only else {},
     )
 
 
 def create_primary_sa_engine() -> sa.AsyncEngine:
     return create_sa_engine(settings.db_dsn_parsed)
 
 
-def create_replica_sa_engine() -> sa.AsyncEngine | None:
-    return create_sa_engine(make_url(settings.db_replica_dsn)) if settings.db_replica_dsn else None
+def create_read_sa_engine() -> sa.AsyncEngine:
+    return create_sa_engine(
+        make_url(settings.db_replica_dsn) if settings.db_replica_dsn else settings.db_dsn_parsed,
+        read_only=True,
+    )
 
 
-async def close_sa_engine(engine: sa.AsyncEngine | None) -> None:
-    if engine:
-        await engine.dispose()
+async def close_sa_engine(engine: sa.AsyncEngine) -> None:
+    await engine.dispose()
 
 
 def choose_sa_engine(
     *,
     primary_engine: sa.AsyncEngine,
-    replica_engine: sa.AsyncEngine | None,
+    read_engine: sa.AsyncEngine,
     request: fastapi.Request | None = None,
 ) -> sa.AsyncEngine:
-    if replica_engine and request and request.method in REPLICA_METHODS:
-        return replica_engine
+    if request and request.method in READ_METHODS:
+        return read_engine
     return primary_engine
 
 

tests/test_db_routing.py

@@ -2,10 +2,12 @@
 
 import fastapi
 import pytest
+import sqlalchemy as sa
 from modern_di import Scope
+from sqlalchemy.exc import DBAPIError
 
-from app import ioc
-from app.resources.db import create_replica_sa_engine, create_sa_engine
+from app import ioc, models
+from app.resources.db import create_read_sa_engine, create_sa_engine
 from app.settings import settings
 
 
@@ -23,9 +25,9 @@ async def primary_engine(di_container: modern_di.Container) -> typing.AsyncItera
 
 
 @pytest.fixture
-async def replica_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
-    engine = create_sa_engine(settings.db_dsn_parsed)
-    di_container.override(ioc.Dependencies.database_replica_engine, engine)
+async def read_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
+    engine = create_read_sa_engine()
+    di_container.override(ioc.Dependencies.database_read_engine, engine)
     yield engine
     await engine.dispose()
 
@@ -37,36 +39,53 @@ def resolve_engine(di_container: modern_di.Container, method: str | None) -> Asy
 
 
 @pytest.mark.parametrize("method", ["GET", "HEAD"])
-def test_safe_methods_use_replica(
-    di_container: modern_di.Container, primary_engine: AsyncEngine, replica_engine: AsyncEngine, method: str
+def test_safe_methods_use_read_engine(
+    di_container: modern_di.Container, primary_engine: AsyncEngine, read_engine: AsyncEngine, method: str
 ) -> None:
     engine = resolve_engine(di_container, method)
-    assert engine is replica_engine
+    assert engine is read_engine
     assert engine is not primary_engine
 
 
 @pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
 def test_write_methods_use_primary(
-    di_container: modern_di.Container, primary_engine: AsyncEngine, replica_engine: AsyncEngine, method: str
+    di_container: modern_di.Container, primary_engine: AsyncEngine, read_engine: AsyncEngine, method: str
 ) -> None:
     engine = resolve_engine(di_container, method)
     assert engine is primary_engine
-    assert engine is not replica_engine
+    assert engine is not read_engine
 
 
-@pytest.mark.usefixtures("replica_engine")
+@pytest.mark.usefixtures("read_engine")
 def test_no_request_uses_primary(di_container: modern_di.Container, primary_engine: AsyncEngine) -> None:
     assert resolve_engine(di_container, None) is primary_engine
 
 
-def test_no_replica_configured_uses_primary(di_container: modern_di.Container, primary_engine: AsyncEngine) -> None:
-    assert di_container.resolve_provider(ioc.Dependencies.database_replica_engine) is None
-    assert resolve_engine(di_container, "GET") is primary_engine
+async def test_read_engine_built_from_primary_dsn_without_replica() -> None:
+    engine = create_read_sa_engine()
+    assert engine.url == settings.db_dsn_parsed
+    await engine.dispose()
 
 
-async def test_replica_engine_built_from_replica_dsn(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_read_engine_built_from_replica_dsn(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(settings, "db_replica_dsn", "postgresql+asyncpg://postgres:password@replica/postgres")
-    engine = create_replica_sa_engine()
-    assert engine
+    engine = create_read_sa_engine()
     assert engine.url.host == "replica"
     await engine.dispose()
+
+
+async def test_read_engine_runs_each_statement_in_its_own_transaction() -> None:
+    engine = create_read_sa_engine()
+    async with engine.connect() as connection:
+        first = await connection.scalar(sa.text("SELECT transaction_timestamp()"))
+        second = await connection.scalar(sa.text("SELECT transaction_timestamp()"))
+    assert first != second
+    await engine.dispose()
+
+
+async def test_read_engine_rejects_writes() -> None:
+    engine = create_read_sa_engine()
+    async with engine.connect() as connection:
+        with pytest.raises(DBAPIError, match="read-only transaction"):
+            await connection.execute(sa.insert(models.Deck).values(name="read only"))
+    await engine.dispose()