feat: route GET/HEAD requests to an optional read replica (#71)

Author: lesnik512

AGENTS.md

@@ -39,8 +39,10 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 ### DI scopes (modern-di)
 
 `app/ioc.py` defines providers:
-- `database_engine` — singleton-ish `AsyncEngine` with `create_sa_engine` / `close_sa_engine` finalizer.
-- `session` — `Scope.REQUEST`, finalized by `close_session`.
+- `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.
+- `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).
 
 Endpoints inject repositories with `FromDI(Repository)` from `modern_di_fastapi`. Add new providers to `Dependencies` rather than constructing services manually in routes.
@@ -49,19 +51,19 @@ Endpoints inject repositories with `FromDI(Repository)` from `modern_di_fastapi`
 
 - `app/models.py` — `BigIntAuditBase` from `advanced_alchemy` (auto `id`, `created_at`, `updated_at`). The module aliases `orm_registry.metadata` onto `orm.DeclarativeBase.metadata` so Alembic autogenerate sees both. New models go here.
 - `app/repositories.py` — Subclass `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. Routes use the service methods (`list`, `get_one_or_none`, `create`, `update`, `create_many`, `upsert_many`).
-- `app/resources/db.py` — `create_session` passes `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the test rollback pattern below: when a test binds the session to a connection already in a transaction, the session owns its own savepoint so the outer transaction survives commits — do not "fix" it.
+- `app/resources/db.py` — engine factories live here and are DI creators: modern-di resolves their parameter annotations at runtime, so `pyproject.toml` exempts `app/resources/` from ruff's `TC` rules. `create_session` passes `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the test rollback pattern below: when a test binds the session to a connection already in a transaction, the session owns its own savepoint so the outer transaction survives commits — do not "fix" it.
 - `migrations/env.py` swaps the asyncpg driver for the sync `postgresql` driver and uses `app.models.METADATA` as `target_metadata`.
 
 ### Settings
 
-`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`.
+`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`.
 
 ### Tests
 
 `tests/conftest.py` provides the test isolation pattern — read it before adding fixtures:
 
 - `app` fixture builds a fresh app via `LifespanManager`.
-- `db_session` opens a connection, begins a transaction, and **overrides `Dependencies.database_engine`** with the connection itself. Each session built against that connection uses `join_transaction_mode="create_savepoint"`, so `auto_commit` releases the session's own savepoint while the outer transaction is rolled back at teardown — each test starts clean.
+- `db_session` opens a connection, begins a transaction, and **overrides `Dependencies.dynamic_engine`** with the connection itself, so reads and writes share one rolled-back transaction. Each session built against that connection uses `join_transaction_mode="create_savepoint"`, so `auto_commit` releases the session's own savepoint while the outer transaction is rolled back at teardown — each test starts clean.
 - `set_async_session_in_base_sqlalchemy_factory` wires `db_session` into `SQLAlchemyFactory.__async_session__` so `polyfactory` factories in `tests/factories.py` (`DeckModelFactory`, `CardModelFactory`) persist via the rolled-back session. Test modules that use these factories opt in with `pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")]`.
 
 `pytest.ini_options` sets `asyncio_mode = "auto"` — async tests do not need `@pytest.mark.asyncio`. Coverage runs by default (`--cov=. --cov-report term-missing`).

app/ioc.py

@@ -1,15 +1,37 @@
 from modern_di import Group, Scope, providers
 
 from app.repositories import CardsRepository, DecksRepository
-from app.resources.db import close_sa_engine, close_session, create_sa_engine, create_session
+from app.resources.db import (
+    choose_sa_engine,
+    close_sa_engine,
+    close_session,
+    create_primary_sa_engine,
+    create_replica_sa_engine,
+    create_session,
+)
 
 
 class Dependencies(Group):
     database_engine = providers.Factory(
-        creator=create_sa_engine, cache=providers.CacheSettings(finalizer=close_sa_engine)
+        creator=create_primary_sa_engine,
+        cache=providers.CacheSettings(finalizer=close_sa_engine),
+        bound_type=None,
+    )
+    database_replica_engine = providers.Factory(
+        creator=create_replica_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},
     )
     session = providers.Factory(
-        scope=Scope.REQUEST, creator=create_session, cache=providers.CacheSettings(finalizer=close_session)
+        scope=Scope.REQUEST,
+        creator=create_session,
+        cache=providers.CacheSettings(finalizer=close_session),
+        kwargs={"engine": dynamic_engine},
     )
 
     decks_repository = providers.Factory(

app/resources/db.py

@@ -2,6 +2,8 @@
 import logging
 import typing
 
+import fastapi
+from sqlalchemy.engine.url import URL, make_url
 from sqlalchemy.ext import asyncio as sa
 
 from app.settings import settings
@@ -10,9 +12,12 @@
 logger = logging.getLogger(__name__)
 
 
-def create_sa_engine() -> sa.AsyncEngine:
+REPLICA_METHODS: typing.Final = frozenset({"GET", "HEAD"})
+
+
+def create_sa_engine(url: URL) -> sa.AsyncEngine:
     return sa.create_async_engine(
-        url=settings.db_dsn_parsed,
+        url=url,
         echo=settings.service_debug,
         echo_pool=settings.service_debug,
         pool_size=settings.db_pool_size,
@@ -21,8 +26,28 @@ def create_sa_engine() -> sa.AsyncEngine:
     )
 
 
-async def close_sa_engine(engine: sa.AsyncEngine) -> None:
-    await engine.dispose()
+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
+
+
+async def close_sa_engine(engine: sa.AsyncEngine | None) -> None:
+    if engine:
+        await engine.dispose()
+
+
+def choose_sa_engine(
+    *,
+    primary_engine: sa.AsyncEngine,
+    replica_engine: sa.AsyncEngine | None,
+    request: fastapi.Request | None = None,
+) -> sa.AsyncEngine:
+    if replica_engine and request and request.method in REPLICA_METHODS:
+        return replica_engine
+    return primary_engine
 
 
 def create_session(engine: sa.AsyncEngine) -> sa.AsyncSession:

app/settings.py

@@ -11,6 +11,7 @@ class Settings(pydantic_settings.BaseSettings):
     log_level: str = "info"
 
     db_dsn: str = "postgresql+asyncpg://postgres:password@db/postgres"
+    db_replica_dsn: str = ""
     db_pool_size: int = 5
     db_max_overflow: int = 0
     db_pool_pre_ping: bool = True

pyproject.toml

@@ -67,6 +67,9 @@ isort.lines-after-imports = 2
 isort.no-lines-before = ["standard-library", "local-folder"]
 
 [tool.ruff.lint.extend-per-file-ignores]
+"app/resources/*.py" = [
+    "TC", # modern-di reads DI creator annotations at runtime
+]
 "tests/*.py" = [
     "S101", # allow asserts
 ]

tests/conftest.py

@@ -9,7 +9,7 @@
 
 from app import ioc
 from app.application import build_app
-from app.resources.db import create_sa_engine
+from app.resources.db import create_primary_sa_engine
 
 
 if typing.TYPE_CHECKING:
@@ -44,10 +44,10 @@ async def di_container(app: fastapi.FastAPI) -> typing.AsyncIterator[modern_di.C
 
 @pytest.fixture
 async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncSession]:
-    engine = create_sa_engine()
+    engine = create_primary_sa_engine()
     connection = await engine.connect()
     transaction = await connection.begin()
-    di_container.override(ioc.Dependencies.database_engine, connection)
+    di_container.override(ioc.Dependencies.dynamic_engine, connection)
 
     try:
         yield AsyncSession(

tests/test_db_routing.py

@@ -0,0 +1,72 @@
+import typing
+
+import fastapi
+import pytest
+from modern_di import Scope
+
+from app import ioc
+from app.resources.db import create_replica_sa_engine, create_sa_engine
+from app.settings import settings
+
+
+if typing.TYPE_CHECKING:
+    import modern_di
+    from sqlalchemy.ext.asyncio import AsyncEngine
+
+
+@pytest.fixture
+async def primary_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
+    engine = create_sa_engine(settings.db_dsn_parsed)
+    di_container.override(ioc.Dependencies.database_engine, engine)
+    yield engine
+    await engine.dispose()
+
+
+@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)
+    yield engine
+    await engine.dispose()
+
+
+def resolve_engine(di_container: modern_di.Container, method: str | None) -> AsyncEngine:
+    context = {fastapi.Request: fastapi.Request({"type": "http", "method": method})} if method else None
+    with di_container.build_child_container(scope=Scope.REQUEST, context=context) as request_container:
+        return request_container.resolve_provider(ioc.Dependencies.dynamic_engine)
+
+
+@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
+) -> None:
+    engine = resolve_engine(di_container, method)
+    assert engine is replica_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
+) -> None:
+    engine = resolve_engine(di_container, method)
+    assert engine is primary_engine
+    assert engine is not replica_engine
+
+
+@pytest.mark.usefixtures("replica_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_replica_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
+    assert engine.url.host == "replica"
+    await engine.dispose()