From 8a1788d81bbf9a841b570d38d2b4797fbbe422b4 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 27 Sep 2026 23:07:22 +0300 Subject: [PATCH] feat: route GET/HEAD requests to an optional read replica --- AGENTS.md | 12 ++++--- app/ioc.py | 28 ++++++++++++++-- app/resources/db.py | 33 +++++++++++++++--- app/settings.py | 1 + pyproject.toml | 3 ++ tests/conftest.py | 6 ++-- tests/test_db_routing.py | 72 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 140 insertions(+), 15 deletions(-) create mode 100644 tests/test_db_routing.py diff --git a/AGENTS.md b/AGENTS.md index 5dce0ba..3b26170 100644 --- a/AGENTS.md +++ b/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`). diff --git a/app/ioc.py b/app/ioc.py index aa1df46..6190593 100644 --- a/app/ioc.py +++ b/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( diff --git a/app/resources/db.py b/app/resources/db.py index 99944bc..3cf5a4d 100644 --- a/app/resources/db.py +++ b/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: diff --git a/app/settings.py b/app/settings.py index 1b28c62..9f16c6f 100644 --- a/app/settings.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 14ba3d0..ed2faf7 100644 --- a/pyproject.toml +++ b/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 ] diff --git a/tests/conftest.py b/tests/conftest.py index 314e500..d4b10b2 100644 --- a/tests/conftest.py +++ b/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( diff --git a/tests/test_db_routing.py b/tests/test_db_routing.py new file mode 100644 index 0000000..700d93a --- /dev/null +++ b/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()