diff --git a/AGENTS.md b/AGENTS.md index b1b0ee2..3b64375 100644 --- a/AGENTS.md +++ b/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). diff --git a/app/ioc.py b/app/ioc.py index 6190593..5cc4224 100644 --- a/app/ioc.py +++ b/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, diff --git a/app/resources/db.py b/app/resources/db.py index 5581697..ab56ace 100644 --- a/app/resources/db.py +++ b/app/resources/db.py @@ -12,10 +12,10 @@ 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, @@ -23,6 +23,8 @@ def create_sa_engine(url: URL) -> sa.AsyncEngine: 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 {}, ) @@ -30,23 +32,25 @@ 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 diff --git a/tests/test_db_routing.py b/tests/test_db_routing.py index 700d93a..a2da494 100644 --- a/tests/test_db_routing.py +++ b/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()