Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).

Expand Down
8 changes: 4 additions & 4 deletions app/ioc.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
close_sa_engine,
close_session,
create_primary_sa_engine,
create_replica_sa_engine,
create_read_sa_engine,
create_session,
)

Expand All @@ -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,
Expand Down
24 changes: 14 additions & 10 deletions app/resources/db.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down
53 changes: 36 additions & 17 deletions tests/test_db_routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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()

Expand All @@ -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()
Loading