perf: turn pool_pre_ping off by default and rely on postgres_retry (#88)

* test: cover handlers against lost connections and the pre-ping default

The handler tests terminate the backend a request session holds, the endpoint tests terminate an idle pooled connection on each engine. Both fail until the handlers retry inside a Transaction and pre-ping is off.

* perf: turn pool_pre_ping off by default and rely on postgres_retry

Every handler is decorated with postgres_retry and runs its queries inside a request-scoped db_retry.Transaction, committing before it leaves the block. Repositories switch to auto_commit=False so the handler owns the whole transaction. Requires db-retry 0.8.0.

* refactor: move each handler's database work into a use case

app/use_cases/ holds one dataclass per operation, retried with postgres_retry and running inside its Transaction, as in chat-app. ioc splits into Database, Repositories and UseCases groups, and handlers only call a use case and convert the result.

* test: replace the per-use-case connection-loss table with one test

Every use case has the same retry and Transaction shape, so ListDecksUseCase stands in for all of them. It runs against a connection lost while held by the session and one lost while idle in the pool. It only reads, so the committed fixture data and the truncation go away.

* test: drop the connection-loss test

db-retry already tests a pooled connection closed while idle being retried through a Transaction against a real Postgres. The template test only ran the same library code through a trivial use case.

Author: lesnik512

AGENTS.md

@@ -30,27 +30,57 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 
 `app/__main__.py` boots Granian pointing at `app.application:build_app` (factory). `build_app()` in `app/application.py`:
 
-1. Creates a `modern_di.Container` with the `Dependencies` group from `app/ioc.py`.
+1. Creates a `modern_di.Container` with `ioc.ALL_GROUPS` from `app/ioc.py`.
 2. Builds a `FastAPIBootstrapper` from `settings.api_bootstrapper_config`, injecting the SQLAlchemy OpenTelemetry instrumentor.
 3. `modern_di_fastapi.setup_di(app, container)` wires DI scopes onto the FastAPI app.
 4. Includes one `ROUTER` per resource (`app.api.decks.ROUTER`, `app.api.cards.ROUTER`) under `/api` via `include_routers`. Add a new resource by creating `app/api/<name>.py` with its own `APIRouter`-based `ROUTER` and registering it there.
 5. Registers `DuplicateKeyError` → 422 handler from `app/exceptions.py`.
 
 ### DI scopes (modern-di)
 
-`app/ioc.py` defines providers:
+`app/ioc.py` splits providers into three groups, collected in `ALL_GROUPS`.
+
+`Database`:
 - `database_engine` — singleton-ish primary `AsyncEngine` (`create_primary_sa_engine`, `close_sa_engine` finalizer).
-- `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.
+- `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 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).
+- `transaction` — `Scope.REQUEST`, a `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session.
+
+`Repositories` (`Scope.REQUEST`): `decks_repository` and `cards_repository`, on `Database.session` with `auto_commit=False`. The use case commits through `transaction.commit()`.
+
+`UseCases` (`Scope.REQUEST`): one provider per use case in `app/use_cases/`. modern-di fills each dataclass field by its type, so a use case needs no `kwargs`.
+
+Endpoints inject a use case with `FromDI(UseCase)` from `modern_di_fastapi`, call it, and convert the result to a schema. They never touch repositories or the session. Add new providers to the matching group rather than constructing services manually in routes.
+
+### Use cases and lost connections
+
+`pool_pre_ping` is off by default (`DB_POOL_PRE_PING`). A pooled connection that died while idle is handled by `db_retry.postgres_retry` instead, which saves a round trip on every checkout. That only works if the retried function owns its whole transaction, so every use case has this shape:
+
+```python
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class ListDecksUseCase:
+    transaction: Transaction
+    decks_repository: DecksRepository
+
+    @postgres_retry
+    async def __call__(self) -> Sequence[models.Deck]:
+        async with self.transaction:
+            decks: typing.Final = await self.decks_repository.get_many()
+            await self.transaction.commit()
+        return decks
+```
+
+- Every query runs inside `async with self.transaction:`. A query before the block leaves the session in a failed transaction on a lost connection, and the retry then fails with `PendingRollbackError`.
+- Commit before leaving the block, reads included, and before any early return. Otherwise `__aexit__` rolls back, which expires the loaded rows.
+- A retry runs the whole use case again. If the connection drops during `COMMIT`, the write may already have landed, so a non-idempotent write such as `CreateDeckUseCase` can be applied twice.
 
-Endpoints inject repositories with `FromDI(Repository)` from `modern_di_fastapi`. Add new providers to `Dependencies` rather than constructing services manually in routes.
+`app/use_cases/` is exempt from ruff's `TC` rules because modern-di reads the dataclass field annotations at runtime.
 
 ### Database layer
 
 - `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/repositories.py` — Subclass `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. Use cases call the service methods (`get_many`, `get_one`, `create`, `update`, `create_many`, `upsert_many`).
 - `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`.
 
@@ -63,7 +93,7 @@ Endpoints inject repositories with `FromDI(Repository)` from `modern_di_fastapi`
 `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.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.
+- `db_session` opens a connection, begins a transaction, and **overrides `Database.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 `transaction.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 with `--cov-fail-under=100`, so `just test` fails unless tests cover every line of new code.

app/api/cards.py

@@ -3,8 +3,11 @@
 import fastapi
 from modern_di_fastapi import FromDI
 
-from app import models, schemas
-from app.repositories import CardsRepository
+from app import schemas
+from app.use_cases.create_cards import CreateCardsUseCase
+from app.use_cases.get_card import GetCardUseCase
+from app.use_cases.list_cards import ListCardsUseCase
+from app.use_cases.update_cards import UpdateCardsUseCase
 
 
 ROUTER: typing.Final = fastapi.APIRouter()
@@ -13,36 +16,32 @@
 @ROUTER.get("/decks/{deck_id}/cards/")
 async def list_cards(
     deck_id: int,
-    cards_repository: CardsRepository = FromDI(CardsRepository),
+    list_cards_use_case: ListCardsUseCase = FromDI(ListCardsUseCase),
 ) -> schemas.Cards:
-    objects = await cards_repository.list_for_deck(deck_id)
-    return schemas.Cards.from_models(objects)
+    return schemas.Cards.from_models(await list_cards_use_case(deck_id=deck_id))
 
 
 @ROUTER.get("/cards/{card_id}/")
 async def get_card(
     card_id: int,
-    cards_repository: CardsRepository = FromDI(CardsRepository),
+    get_card_use_case: GetCardUseCase = FromDI(GetCardUseCase),
 ) -> schemas.Card:
-    instance = await cards_repository.get_one(models.Card.id == card_id)
-    return schemas.Card.model_validate(instance)
+    return schemas.Card.model_validate(await get_card_use_case(card_id=card_id))
 
 
 @ROUTER.post("/decks/{deck_id}/cards/")
 async def create_cards(
     deck_id: int,
     data: list[schemas.CardCreate],
-    cards_repository: CardsRepository = FromDI(CardsRepository),
+    create_cards_use_case: CreateCardsUseCase = FromDI(CreateCardsUseCase),
 ) -> schemas.Cards:
-    objects = await cards_repository.add_cards(deck_id, data)
-    return schemas.Cards.from_models(objects)
+    return schemas.Cards.from_models(await create_cards_use_case(deck_id=deck_id, data=data))
 
 
 @ROUTER.put("/decks/{deck_id}/cards/")
 async def update_cards(
     deck_id: int,
     data: list[schemas.Card],
-    cards_repository: CardsRepository = FromDI(CardsRepository),
+    update_cards_use_case: UpdateCardsUseCase = FromDI(UpdateCardsUseCase),
 ) -> schemas.Cards:
-    objects = await cards_repository.upsert_cards(deck_id, data)
-    return schemas.Cards.from_models(objects)
+    return schemas.Cards.from_models(await update_cards_use_case(deck_id=deck_id, data=data))

app/api/decks.py

@@ -4,43 +4,42 @@
 from modern_di_fastapi import FromDI
 
 from app import schemas
-from app.repositories import DecksRepository
+from app.use_cases.create_deck import CreateDeckUseCase
+from app.use_cases.get_deck import GetDeckUseCase
+from app.use_cases.list_decks import ListDecksUseCase
+from app.use_cases.update_deck import UpdateDeckUseCase
 
 
 ROUTER: typing.Final = fastapi.APIRouter()
 
 
 @ROUTER.get("/decks/")
 async def list_decks(
-    decks_repository: DecksRepository = FromDI(DecksRepository),
+    list_decks_use_case: ListDecksUseCase = FromDI(ListDecksUseCase),
 ) -> schemas.Decks:
-    objects = await decks_repository.get_many()
-    return schemas.Decks.from_models(objects)
+    return schemas.Decks.from_models(await list_decks_use_case())
 
 
 @ROUTER.get("/decks/{deck_id}/")
 async def get_deck(
     deck_id: int,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
+    get_deck_use_case: GetDeckUseCase = FromDI(GetDeckUseCase),
 ) -> schemas.DeckWithCards:
-    instance = await decks_repository.fetch_with_cards(deck_id)
-    return schemas.DeckWithCards.model_validate(instance)
+    return schemas.DeckWithCards.model_validate(await get_deck_use_case(deck_id=deck_id))
 
 
 @ROUTER.put("/decks/{deck_id}/")
 async def update_deck(
     deck_id: int,
     data: schemas.DeckCreate,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
+    update_deck_use_case: UpdateDeckUseCase = FromDI(UpdateDeckUseCase),
 ) -> schemas.Deck:
-    instance = await decks_repository.update(data=data.model_dump(), item_id=deck_id)
-    return schemas.Deck.model_validate(instance)
+    return schemas.Deck.model_validate(await update_deck_use_case(deck_id=deck_id, data=data))
 
 
 @ROUTER.post("/decks/")
 async def create_deck(
     data: schemas.DeckCreate,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
+    create_deck_use_case: CreateDeckUseCase = FromDI(CreateDeckUseCase),
 ) -> schemas.Deck:
-    instance = await decks_repository.create(data.model_dump())
-    return schemas.Deck.model_validate(instance)
+    return schemas.Deck.model_validate(await create_deck_use_case(data=data))

app/application.py

@@ -22,7 +22,7 @@ def include_routers(app: fastapi.FastAPI) -> None:
 
 
 def build_app() -> fastapi.FastAPI:
-    di_container = modern_di.Container(groups=[ioc.Dependencies])
+    di_container = modern_di.Container(groups=ioc.ALL_GROUPS)
     bootstrap_config = dataclasses.replace(
         settings.api_bootstrapper_config,
         opentelemetry_instrumentors=[SQLAlchemyInstrumentor()],

app/ioc.py

@@ -1,3 +1,6 @@
+import typing
+
+from db_retry import Transaction
 from modern_di import Group, Scope, providers
 
 from app.repositories import CardsRepository, DecksRepository
@@ -9,9 +12,17 @@
     create_read_sa_engine,
     create_session,
 )
+from app.use_cases.create_cards import CreateCardsUseCase
+from app.use_cases.create_deck import CreateDeckUseCase
+from app.use_cases.get_card import GetCardUseCase
+from app.use_cases.get_deck import GetDeckUseCase
+from app.use_cases.list_cards import ListCardsUseCase
+from app.use_cases.list_decks import ListDecksUseCase
+from app.use_cases.update_cards import UpdateCardsUseCase
+from app.use_cases.update_deck import UpdateDeckUseCase
 
 
-class Dependencies(Group):
+class Database(Group):
     database_engine = providers.Factory(
         creator=create_primary_sa_engine,
         cache=providers.CacheSettings(finalizer=close_sa_engine),
@@ -33,14 +44,33 @@ class Dependencies(Group):
         cache=providers.CacheSettings(finalizer=close_session),
         kwargs={"engine": dynamic_engine},
     )
+    transaction = providers.Factory(
+        scope=Scope.REQUEST,
+        creator=Transaction,
+        kwargs={"session": session},
+    )
 
+
+class Repositories(Group, scope=Scope.REQUEST):
     decks_repository = providers.Factory(
-        scope=Scope.REQUEST,
         creator=DecksRepository,
-        kwargs={"auto_commit": True, "session": session},
+        kwargs={"session": Database.session, "auto_commit": False},
     )
     cards_repository = providers.Factory(
-        scope=Scope.REQUEST,
         creator=CardsRepository,
-        kwargs={"auto_commit": True, "session": session},
+        kwargs={"session": Database.session, "auto_commit": False},
     )
+
+
+class UseCases(Group, scope=Scope.REQUEST):
+    list_decks_use_case = providers.Factory(creator=ListDecksUseCase)
+    get_deck_use_case = providers.Factory(creator=GetDeckUseCase)
+    update_deck_use_case = providers.Factory(creator=UpdateDeckUseCase)
+    create_deck_use_case = providers.Factory(creator=CreateDeckUseCase)
+    list_cards_use_case = providers.Factory(creator=ListCardsUseCase)
+    get_card_use_case = providers.Factory(creator=GetCardUseCase)
+    create_cards_use_case = providers.Factory(creator=CreateCardsUseCase)
+    update_cards_use_case = providers.Factory(creator=UpdateCardsUseCase)
+
+
+ALL_GROUPS: typing.Final[list[type[Group]]] = [Database, Repositories, UseCases]

app/settings.py

@@ -14,7 +14,7 @@ class Settings(pydantic_settings.BaseSettings):
     db_replica_dsn: str = ""
     db_pool_size: int = 5
     db_max_overflow: int = 0
-    db_pool_pre_ping: bool = True
+    db_pool_pre_ping: bool = False
 
     app_host: str = "0.0.0.0"  # noqa: S104
     app_port: int = 8000

app/use_cases/__init__.py

@@ -0,0 +1,21 @@
+import dataclasses
+import typing
+from collections.abc import Sequence
+
+from db_retry import Transaction, postgres_retry
+
+from app import models, schemas
+from app.repositories import CardsRepository
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class CreateCardsUseCase:
+    transaction: Transaction
+    cards_repository: CardsRepository
+
+    @postgres_retry
+    async def __call__(self, *, deck_id: int, data: list[schemas.CardCreate]) -> Sequence[models.Card]:
+        async with self.transaction:
+            cards: typing.Final = await self.cards_repository.add_cards(deck_id, data)
+            await self.transaction.commit()
+        return cards

app/use_cases/create_cards.py

@@ -0,0 +1,20 @@
+import dataclasses
+import typing
+
+from db_retry import Transaction, postgres_retry
+
+from app import models, schemas
+from app.repositories import DecksRepository
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class CreateDeckUseCase:
+    transaction: Transaction
+    decks_repository: DecksRepository
+
+    @postgres_retry
+    async def __call__(self, *, data: schemas.DeckCreate) -> models.Deck:
+        async with self.transaction:
+            deck: typing.Final = await self.decks_repository.create(data.model_dump())
+            await self.transaction.commit()
+        return deck

app/use_cases/create_deck.py

@@ -0,0 +1,20 @@
+import dataclasses
+import typing
+
+from db_retry import Transaction, postgres_retry
+
+from app import models
+from app.repositories import CardsRepository
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class GetCardUseCase:
+    transaction: Transaction
+    cards_repository: CardsRepository
+
+    @postgres_retry
+    async def __call__(self, *, card_id: int) -> models.Card:
+        async with self.transaction:
+            card: typing.Final = await self.cards_repository.get_one(models.Card.id == card_id)
+            await self.transaction.commit()
+        return card