From 14c1b36fdd502ae42a41e04bcad3b3e8733f8652 Mon Sep 17 00:00:00 2001
From: Artur Shiriev <lesnik512@gmail.com>
Date: Fri, 16 Oct 2026 21:24:20 +0300
Subject: [PATCH 1/5] 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.
---
 tests/test_connection_loss.py | 164 ++++++++++++++++++++++++++++++++++
 tests/test_settings.py        |   4 +
 2 files changed, 168 insertions(+)
 create mode 100644 tests/test_connection_loss.py

diff --git a/tests/test_connection_loss.py b/tests/test_connection_loss.py
new file mode 100644
index 0000000..aebe4c3
--- /dev/null
+++ b/tests/test_connection_loss.py
@@ -0,0 +1,164 @@
+import asyncio
+import inspect
+import typing
+
+import asyncpg
+import modern_di
+import pytest
+import sqlalchemy as sa
+from fastapi import status
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession
+
+from app import ioc, models, schemas
+from app.api import cards, decks
+from app.resources.db import create_primary_sa_engine, create_read_sa_engine
+
+
+if typing.TYPE_CHECKING:
+    from httpx import AsyncClient
+
+
+class StoredIds(typing.NamedTuple):
+    deck_id: int
+    card_id: int
+
+
+async def _truncate(engine: AsyncEngine) -> None:
+    async with engine.begin() as connection:
+        await connection.execute(sa.text("TRUNCATE cards, decks RESTART IDENTITY"))
+
+
+@pytest.fixture
+async def stored_ids() -> typing.AsyncIterator[StoredIds]:
+    engine: typing.Final = create_primary_sa_engine()
+    await _truncate(engine)
+    async with engine.begin() as connection:
+        deck_id: typing.Final = (
+            await connection.execute(sa.insert(models.Deck).values(name="deck").returning(models.Deck.id))
+        ).scalar_one()
+        card_id: typing.Final = (
+            await connection.execute(
+                sa.insert(models.Card).values(front="front", deck_id=deck_id).returning(models.Card.id)
+            )
+        ).scalar_one()
+    try:
+        yield StoredIds(deck_id=deck_id, card_id=card_id)
+    finally:
+        await _truncate(engine)
+        await engine.dispose()
+
+
+@pytest.fixture
+async def engines(di_container: modern_di.Container) -> typing.AsyncIterator[tuple[AsyncEngine, AsyncEngine]]:
+    primary_engine: typing.Final = create_primary_sa_engine()
+    read_engine: typing.Final = create_read_sa_engine()
+    di_container.override(ioc.Dependencies.database_engine, primary_engine)
+    di_container.override(ioc.Dependencies.database_read_engine, read_engine)
+    try:
+        yield primary_engine, read_engine
+    finally:
+        di_container.reset_override()
+        await primary_engine.dispose()
+        await read_engine.dispose()
+
+
+@pytest.fixture
+async def request_container(di_container: modern_di.Container) -> typing.AsyncIterator[modern_di.Container]:
+    async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
+        yield container
+
+
+async def _watch_backend(connection: AsyncConnection) -> tuple[int, asyncio.Event]:
+    pid: typing.Final = await connection.scalar(sa.text("SELECT pg_backend_pid()"))
+    driver_connection: typing.Final = (await connection.get_raw_connection()).driver_connection
+    assert driver_connection is not None
+    closed: typing.Final = asyncio.Event()
+    driver_connection.add_termination_listener(lambda _: closed.set())
+    return pid, closed
+
+
+async def _terminate_backend(engine: AsyncEngine, pid: int, closed: asyncio.Event) -> None:
+    killer: typing.Final = await asyncpg.connect(
+        engine.url.set(drivername="postgresql").render_as_string(hide_password=False)
+    )
+    try:
+        await killer.execute("SELECT pg_terminate_backend($1)", pid)
+    finally:
+        await killer.close()
+    await asyncio.wait_for(closed.wait(), timeout=5)
+
+
+@pytest.fixture
+async def lost_connection(engines: tuple[AsyncEngine, AsyncEngine], request_container: modern_di.Container) -> None:
+    """Leave the request session holding a primary connection whose backend was terminated."""
+    session: typing.Final = request_container.resolve_provider(ioc.Dependencies.session)
+    assert isinstance(session, AsyncSession)
+    connection: typing.Final = await session.connection()
+    primary_engine, _ = engines
+    await _terminate_backend(primary_engine, *await _watch_backend(connection))
+
+
+@pytest.fixture
+async def lost_pooled_connection(engines: tuple[AsyncEngine, AsyncEngine]) -> None:
+    """Leave each engine's pool holding one idle connection whose backend was terminated."""
+    for engine in engines:
+        async with engine.connect() as connection:
+            watched = await _watch_backend(connection)
+        await _terminate_backend(engine, *watched)
+
+
+@pytest.mark.usefixtures("lost_pooled_connection")
+@pytest.mark.parametrize(
+    "build_request",
+    [
+        lambda _: ("GET", "/api/decks/", None),
+        lambda ids: ("GET", f"/api/decks/{ids.deck_id}/", None),
+        lambda ids: ("PUT", f"/api/decks/{ids.deck_id}/", {"name": "renamed"}),
+        lambda _: ("POST", "/api/decks/", {"name": "new"}),
+        lambda ids: ("GET", f"/api/decks/{ids.deck_id}/cards/", None),
+        lambda ids: ("GET", f"/api/cards/{ids.card_id}/", None),
+        lambda ids: ("POST", f"/api/decks/{ids.deck_id}/cards/", [{"front": "new"}]),
+        lambda ids: ("PUT", f"/api/decks/{ids.deck_id}/cards/", [{"id": ids.card_id, "front": "renamed"}]),
+    ],
+)
+async def test_endpoint_recovers_from_a_connection_lost_while_pooled(
+    client: AsyncClient,
+    stored_ids: StoredIds,
+    build_request: typing.Callable[[StoredIds], tuple[str, str, object]],
+) -> None:
+    method, path, json = build_request(stored_ids)
+    response: typing.Final = await client.request(method, path, json=json)
+    assert response.status_code == status.HTTP_200_OK
+
+
+@pytest.mark.usefixtures("lost_connection")
+@pytest.mark.parametrize(
+    ("handler", "build_kwargs"),
+    [
+        (decks.list_decks, lambda _: {}),
+        (decks.get_deck, lambda ids: {"deck_id": ids.deck_id}),
+        (decks.update_deck, lambda ids: {"deck_id": ids.deck_id, "data": schemas.DeckCreate(name="renamed")}),
+        (decks.create_deck, lambda _: {"data": schemas.DeckCreate(name="new")}),
+        (cards.list_cards, lambda ids: {"deck_id": ids.deck_id}),
+        (cards.get_card, lambda ids: {"card_id": ids.card_id}),
+        (cards.create_cards, lambda ids: {"deck_id": ids.deck_id, "data": [schemas.CardCreate(front="new")]}),
+        (
+            cards.update_cards,
+            lambda ids: {"deck_id": ids.deck_id, "data": [schemas.Card(id=ids.card_id, front="renamed")]},
+        ),
+    ],
+)
+async def test_handler_recovers_from_a_lost_connection(
+    request_container: modern_di.Container,
+    stored_ids: StoredIds,
+    handler: typing.Callable[..., typing.Awaitable[schemas.Base]],
+    build_kwargs: typing.Callable[[StoredIds], dict[str, typing.Any]],
+) -> None:
+    dependencies: typing.Final = {
+        "decks_repository": request_container.resolve_provider(ioc.Dependencies.decks_repository),
+        "cards_repository": request_container.resolve_provider(ioc.Dependencies.cards_repository),
+        "transaction": request_container.resolve_provider(ioc.Dependencies.transaction),
+    }
+    parameters: typing.Final = inspect.signature(handler).parameters
+    kwargs: typing.Final = {name: value for name, value in dependencies.items() if name in parameters}
+    assert isinstance(await handler(**kwargs, **build_kwargs(stored_ids)), schemas.Base)
diff --git a/tests/test_settings.py b/tests/test_settings.py
index 48949de..f1a9573 100644
--- a/tests/test_settings.py
+++ b/tests/test_settings.py
@@ -8,3 +8,7 @@ def test_api_bootstrapper_config_reads_instance_not_global() -> None:
 
     assert config.service_name == "custom-service"
     assert config.service_version == "9.9.9"
+
+
+def test_pool_pre_ping_is_off_by_default() -> None:
+    assert Settings().db_pool_pre_ping is False

From 2b97e89bc60873bd9b32d169f3f47282df95e48e Mon Sep 17 00:00:00 2001
From: Artur Shiriev <lesnik512@gmail.com>
Date: Fri, 16 Oct 2026 21:24:20 +0300
Subject: [PATCH 2/5] 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.
---
 AGENTS.md        | 33 +++++++++++++++++++++++++++++----
 app/api/cards.py | 25 +++++++++++++++++++++----
 app/api/decks.py | 25 +++++++++++++++++++++----
 app/ioc.py       | 10 ++++++++--
 app/settings.py  |  2 +-
 pyproject.toml   |  1 +
 uv.lock          | 15 +++++++++++++++
 7 files changed, 96 insertions(+), 15 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index 3b64375..ba49f53 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -40,12 +40,35 @@ 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_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.
+- `decks_repository`, `cards_repository` — `Scope.REQUEST`, depend on `session`, configured with `auto_commit=False`. The handler commits through `transaction.commit()`.
+
+Endpoints inject repositories and `Transaction` with `FromDI(...)` from `modern_di_fastapi`. Add new providers to `Dependencies` rather than constructing services manually in routes.
+
+### 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 handler that touches the database has this shape:
+
+```python
+@ROUTER.get("/decks/")
+@postgres_retry
+async def list_decks(
+    decks_repository: DecksRepository = FromDI(DecksRepository),
+    transaction: Transaction = FromDI(Transaction),
+) -> schemas.Decks:
+    async with transaction:
+        objects = await decks_repository.get_many()
+        await transaction.commit()
+    return schemas.Decks.from_models(objects)
+```
 
-Endpoints inject repositories with `FromDI(Repository)` from `modern_di_fastapi`. Add new providers to `Dependencies` rather than constructing services manually in routes.
+- `@postgres_retry` goes under the route decorator.
+- Every query runs inside `async with 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 handler again. If the connection drops during `COMMIT`, the write may already have landed, so a non-idempotent write such as `create_deck` can be applied twice.
 
 ### Database layer
 
@@ -63,9 +86,11 @@ 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 `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 `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")]`.
 
+`tests/test_connection_loss.py` overrides both engines with fresh ones and terminates a backend with `pg_terminate_backend`: `lost_pooled_connection` kills an idle pooled connection on each engine and goes through the HTTP client, `lost_connection` kills the connection a request session already holds and calls the handler directly. These tests commit for real, so `stored_ids` truncates `decks` and `cards` before and after. Each retry sleeps 1 to 2 seconds, which is most of the suite's run time.
+
 `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.
 
 ## Conventions
diff --git a/app/api/cards.py b/app/api/cards.py
index 0291b3c..0f80c05 100644
--- a/app/api/cards.py
+++ b/app/api/cards.py
@@ -1,6 +1,7 @@
 import typing
 
 import fastapi
+from db_retry import Transaction, postgres_retry
 from modern_di_fastapi import FromDI
 
 from app import models, schemas
@@ -11,38 +12,54 @@
 
 
 @ROUTER.get("/decks/{deck_id}/cards/")
+@postgres_retry
 async def list_cards(
     deck_id: int,
     cards_repository: CardsRepository = FromDI(CardsRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Cards:
-    objects = await cards_repository.list_for_deck(deck_id)
+    async with transaction:
+        objects = await cards_repository.list_for_deck(deck_id)
+        await transaction.commit()
     return schemas.Cards.from_models(objects)
 
 
 @ROUTER.get("/cards/{card_id}/")
+@postgres_retry
 async def get_card(
     card_id: int,
     cards_repository: CardsRepository = FromDI(CardsRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Card:
-    instance = await cards_repository.get_one(models.Card.id == card_id)
+    async with transaction:
+        instance = await cards_repository.get_one(models.Card.id == card_id)
+        await transaction.commit()
     return schemas.Card.model_validate(instance)
 
 
 @ROUTER.post("/decks/{deck_id}/cards/")
+@postgres_retry
 async def create_cards(
     deck_id: int,
     data: list[schemas.CardCreate],
     cards_repository: CardsRepository = FromDI(CardsRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Cards:
-    objects = await cards_repository.add_cards(deck_id, data)
+    async with transaction:
+        objects = await cards_repository.add_cards(deck_id, data)
+        await transaction.commit()
     return schemas.Cards.from_models(objects)
 
 
 @ROUTER.put("/decks/{deck_id}/cards/")
+@postgres_retry
 async def update_cards(
     deck_id: int,
     data: list[schemas.Card],
     cards_repository: CardsRepository = FromDI(CardsRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Cards:
-    objects = await cards_repository.upsert_cards(deck_id, data)
+    async with transaction:
+        objects = await cards_repository.upsert_cards(deck_id, data)
+        await transaction.commit()
     return schemas.Cards.from_models(objects)
diff --git a/app/api/decks.py b/app/api/decks.py
index de0e465..2238d77 100644
--- a/app/api/decks.py
+++ b/app/api/decks.py
@@ -1,6 +1,7 @@
 import typing
 
 import fastapi
+from db_retry import Transaction, postgres_retry
 from modern_di_fastapi import FromDI
 
 from app import schemas
@@ -11,36 +12,52 @@
 
 
 @ROUTER.get("/decks/")
+@postgres_retry
 async def list_decks(
     decks_repository: DecksRepository = FromDI(DecksRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Decks:
-    objects = await decks_repository.get_many()
+    async with transaction:
+        objects = await decks_repository.get_many()
+        await transaction.commit()
     return schemas.Decks.from_models(objects)
 
 
 @ROUTER.get("/decks/{deck_id}/")
+@postgres_retry
 async def get_deck(
     deck_id: int,
     decks_repository: DecksRepository = FromDI(DecksRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.DeckWithCards:
-    instance = await decks_repository.fetch_with_cards(deck_id)
+    async with transaction:
+        instance = await decks_repository.fetch_with_cards(deck_id)
+        await transaction.commit()
     return schemas.DeckWithCards.model_validate(instance)
 
 
 @ROUTER.put("/decks/{deck_id}/")
+@postgres_retry
 async def update_deck(
     deck_id: int,
     data: schemas.DeckCreate,
     decks_repository: DecksRepository = FromDI(DecksRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Deck:
-    instance = await decks_repository.update(data=data.model_dump(), item_id=deck_id)
+    async with transaction:
+        instance = await decks_repository.update(data=data.model_dump(), item_id=deck_id)
+        await transaction.commit()
     return schemas.Deck.model_validate(instance)
 
 
 @ROUTER.post("/decks/")
+@postgres_retry
 async def create_deck(
     data: schemas.DeckCreate,
     decks_repository: DecksRepository = FromDI(DecksRepository),
+    transaction: Transaction = FromDI(Transaction),
 ) -> schemas.Deck:
-    instance = await decks_repository.create(data.model_dump())
+    async with transaction:
+        instance = await decks_repository.create(data.model_dump())
+        await transaction.commit()
     return schemas.Deck.model_validate(instance)
diff --git a/app/ioc.py b/app/ioc.py
index 5cc4224..6bc0e73 100644
--- a/app/ioc.py
+++ b/app/ioc.py
@@ -1,3 +1,4 @@
+from db_retry import Transaction
 from modern_di import Group, Scope, providers
 
 from app.repositories import CardsRepository, DecksRepository
@@ -33,14 +34,19 @@ class Dependencies(Group):
         cache=providers.CacheSettings(finalizer=close_session),
         kwargs={"engine": dynamic_engine},
     )
+    transaction = providers.Factory(
+        scope=Scope.REQUEST,
+        creator=Transaction,
+        kwargs={"session": session},
+    )
 
     decks_repository = providers.Factory(
         scope=Scope.REQUEST,
         creator=DecksRepository,
-        kwargs={"auto_commit": True, "session": session},
+        kwargs={"auto_commit": False, "session": session},
     )
     cards_repository = providers.Factory(
         scope=Scope.REQUEST,
         creator=CardsRepository,
-        kwargs={"auto_commit": True, "session": session},
+        kwargs={"auto_commit": False, "session": session},
     )
diff --git a/app/settings.py b/app/settings.py
index 5b87b5d..978764a 100644
--- a/app/settings.py
+++ b/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
diff --git a/pyproject.toml b/pyproject.toml
index a23c6ac..0f9939a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -20,6 +20,7 @@ dependencies = [
     "psycopg",
     "sqlalchemy[asyncio]<2.1",
     "asyncpg",
+    "db-retry>=0.8.0",
     # tracing
     "opentelemetry-instrumentation-sqlalchemy",
 ]
diff --git a/uv.lock b/uv.lock
index 4ea4a91..9788809 100644
--- a/uv.lock
+++ b/uv.lock
@@ -203,6 +203,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/3f/0c/7a64e1ac90541a8edf50daef0914848011fb057a5bf55284a4811e21939a/coverage-7.16.2-py3-none-any.whl", hash = "sha256:11d28e9123a9156cb405d8d27b44256c9a58fb5decc2073a8f17862057e3aa0f", size = 215754, upload-time = "2026-09-27T12:28:59.075Z" },
 ]
 
+[[package]]
+name = "db-retry"
+version = "0.8.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "asyncpg" },
+    { name = "sqlalchemy", extra = ["asyncio"] },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/cb/78/71a71ccabcd09175e4351c655b0d1439a6207f9401ce8df4189fa8bfcc84/db_retry-0.8.0.tar.gz", hash = "sha256:dccdc281422f8d01acc2b4f39e622d4a97e03b23adf3d3bf66b61c680c235b71", size = 8737, upload-time = "2026-10-16T15:55:58.283Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/d0/1c/3becec5f059559f88c2b4e343673509bac363ebb255743d144da7af35f95/db_retry-0.8.0-py3-none-any.whl", hash = "sha256:edaf2aed767082eb84a6d22b8d79001d0f2c86d68853dda46a623c94fe819ad3", size = 9338, upload-time = "2026-10-16T15:55:57.339Z" },
+]
+
 [[package]]
 name = "eof-fixer"
 version = "0.4.3"
@@ -251,6 +264,7 @@ dependencies = [
     { name = "advanced-alchemy" },
     { name = "alembic" },
     { name = "asyncpg" },
+    { name = "db-retry" },
     { name = "fastapi" },
     { name = "granian", extra = ["uvloop"] },
     { name = "lite-bootstrap", extra = ["fastapi-all"] },
@@ -281,6 +295,7 @@ requires-dist = [
     { name = "advanced-alchemy" },
     { name = "alembic" },
     { name = "asyncpg" },
+    { name = "db-retry", specifier = ">=0.8.0" },
     { name = "fastapi", specifier = ">=0.76" },
     { name = "granian", extras = ["uvloop"] },
     { name = "lite-bootstrap", extras = ["fastapi-all"] },

From d73a36d04ea144a74b92644de67422cbd5c033f0 Mon Sep 17 00:00:00 2001
From: Artur Shiriev <lesnik512@gmail.com>
Date: Fri, 16 Oct 2026 21:53:32 +0300
Subject: [PATCH 3/5] 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.
---
 AGENTS.md                     | 51 ++++++++++++++++++++---------------
 app/api/cards.py              | 44 +++++++++---------------------
 app/api/decks.py              | 42 +++++++++--------------------
 app/application.py            |  2 +-
 app/ioc.py                    | 34 +++++++++++++++++++----
 app/use_cases/__init__.py     |  0
 app/use_cases/create_cards.py | 21 +++++++++++++++
 app/use_cases/create_deck.py  | 20 ++++++++++++++
 app/use_cases/get_card.py     | 20 ++++++++++++++
 app/use_cases/get_deck.py     | 20 ++++++++++++++
 app/use_cases/list_cards.py   | 21 +++++++++++++++
 app/use_cases/list_decks.py   | 21 +++++++++++++++
 app/use_cases/update_cards.py | 21 +++++++++++++++
 app/use_cases/update_deck.py  | 20 ++++++++++++++
 pyproject.toml                |  3 +++
 tests/conftest.py             |  2 +-
 tests/test_connection_loss.py | 46 +++++++++++++++----------------
 tests/test_db_routing.py      |  6 ++---
 tests/test_main.py            |  2 +-
 19 files changed, 278 insertions(+), 118 deletions(-)
 create mode 100644 app/use_cases/__init__.py
 create mode 100644 app/use_cases/create_cards.py
 create mode 100644 app/use_cases/create_deck.py
 create mode 100644 app/use_cases/get_card.py
 create mode 100644 app/use_cases/get_deck.py
 create mode 100644 app/use_cases/list_cards.py
 create mode 100644 app/use_cases/list_decks.py
 create mode 100644 app/use_cases/update_cards.py
 create mode 100644 app/use_cases/update_deck.py

diff --git a/AGENTS.md b/AGENTS.md
index ba49f53..3a71615 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -30,7 +30,7 @@ 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.
@@ -38,42 +38,49 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 
 ### 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 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`.
 - `transaction` — `Scope.REQUEST`, a `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session.
-- `decks_repository`, `cards_repository` — `Scope.REQUEST`, depend on `session`, configured with `auto_commit=False`. The handler commits through `transaction.commit()`.
 
-Endpoints inject repositories and `Transaction` with `FromDI(...)` from `modern_di_fastapi`. Add new providers to `Dependencies` rather than constructing services manually in routes.
+`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`.
 
-### Lost connections
+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.
 
-`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 handler that touches the database has this shape:
+### 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
-@ROUTER.get("/decks/")
-@postgres_retry
-async def list_decks(
-    decks_repository: DecksRepository = FromDI(DecksRepository),
-    transaction: Transaction = FromDI(Transaction),
-) -> schemas.Decks:
-    async with transaction:
-        objects = await decks_repository.get_many()
-        await transaction.commit()
-    return schemas.Decks.from_models(objects)
+@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
 ```
 
-- `@postgres_retry` goes under the route decorator.
-- Every query runs inside `async with transaction:`. A query before the block leaves the session in a failed transaction on a lost connection, and the retry then fails with `PendingRollbackError`.
+- 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 handler again. If the connection drops during `COMMIT`, the write may already have landed, so a non-idempotent write such as `create_deck` can be applied twice.
+- 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.
+
+`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`.
 
@@ -86,10 +93,10 @@ async def list_decks(
 `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 `transaction.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")]`.
 
-`tests/test_connection_loss.py` overrides both engines with fresh ones and terminates a backend with `pg_terminate_backend`: `lost_pooled_connection` kills an idle pooled connection on each engine and goes through the HTTP client, `lost_connection` kills the connection a request session already holds and calls the handler directly. These tests commit for real, so `stored_ids` truncates `decks` and `cards` before and after. Each retry sleeps 1 to 2 seconds, which is most of the suite's run time.
+`tests/test_connection_loss.py` overrides both engines with fresh ones and terminates a backend with `pg_terminate_backend`: `lost_pooled_connection` kills an idle pooled connection on each engine and goes through the HTTP client, `lost_connection` kills the connection a request session already holds and calls each use case directly. These tests commit for real, so `stored_ids` truncates `decks` and `cards` before and after. Each retry sleeps 1 to 2 seconds, which is most of the suite's run time.
 
 `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.
 
diff --git a/app/api/cards.py b/app/api/cards.py
index 0f80c05..36f08cd 100644
--- a/app/api/cards.py
+++ b/app/api/cards.py
@@ -1,65 +1,47 @@
 import typing
 
 import fastapi
-from db_retry import Transaction, postgres_retry
 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()
 
 
 @ROUTER.get("/decks/{deck_id}/cards/")
-@postgres_retry
 async def list_cards(
     deck_id: int,
-    cards_repository: CardsRepository = FromDI(CardsRepository),
-    transaction: Transaction = FromDI(Transaction),
+    list_cards_use_case: ListCardsUseCase = FromDI(ListCardsUseCase),
 ) -> schemas.Cards:
-    async with transaction:
-        objects = await cards_repository.list_for_deck(deck_id)
-        await transaction.commit()
-    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}/")
-@postgres_retry
 async def get_card(
     card_id: int,
-    cards_repository: CardsRepository = FromDI(CardsRepository),
-    transaction: Transaction = FromDI(Transaction),
+    get_card_use_case: GetCardUseCase = FromDI(GetCardUseCase),
 ) -> schemas.Card:
-    async with transaction:
-        instance = await cards_repository.get_one(models.Card.id == card_id)
-        await transaction.commit()
-    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/")
-@postgres_retry
 async def create_cards(
     deck_id: int,
     data: list[schemas.CardCreate],
-    cards_repository: CardsRepository = FromDI(CardsRepository),
-    transaction: Transaction = FromDI(Transaction),
+    create_cards_use_case: CreateCardsUseCase = FromDI(CreateCardsUseCase),
 ) -> schemas.Cards:
-    async with transaction:
-        objects = await cards_repository.add_cards(deck_id, data)
-        await transaction.commit()
-    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/")
-@postgres_retry
 async def update_cards(
     deck_id: int,
     data: list[schemas.Card],
-    cards_repository: CardsRepository = FromDI(CardsRepository),
-    transaction: Transaction = FromDI(Transaction),
+    update_cards_use_case: UpdateCardsUseCase = FromDI(UpdateCardsUseCase),
 ) -> schemas.Cards:
-    async with transaction:
-        objects = await cards_repository.upsert_cards(deck_id, data)
-        await transaction.commit()
-    return schemas.Cards.from_models(objects)
+    return schemas.Cards.from_models(await update_cards_use_case(deck_id=deck_id, data=data))
diff --git a/app/api/decks.py b/app/api/decks.py
index 2238d77..29eb8f2 100644
--- a/app/api/decks.py
+++ b/app/api/decks.py
@@ -1,63 +1,45 @@
 import typing
 
 import fastapi
-from db_retry import Transaction, postgres_retry
 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/")
-@postgres_retry
 async def list_decks(
-    decks_repository: DecksRepository = FromDI(DecksRepository),
-    transaction: Transaction = FromDI(Transaction),
+    list_decks_use_case: ListDecksUseCase = FromDI(ListDecksUseCase),
 ) -> schemas.Decks:
-    async with transaction:
-        objects = await decks_repository.get_many()
-        await transaction.commit()
-    return schemas.Decks.from_models(objects)
+    return schemas.Decks.from_models(await list_decks_use_case())
 
 
 @ROUTER.get("/decks/{deck_id}/")
-@postgres_retry
 async def get_deck(
     deck_id: int,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
-    transaction: Transaction = FromDI(Transaction),
+    get_deck_use_case: GetDeckUseCase = FromDI(GetDeckUseCase),
 ) -> schemas.DeckWithCards:
-    async with transaction:
-        instance = await decks_repository.fetch_with_cards(deck_id)
-        await transaction.commit()
-    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}/")
-@postgres_retry
 async def update_deck(
     deck_id: int,
     data: schemas.DeckCreate,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
-    transaction: Transaction = FromDI(Transaction),
+    update_deck_use_case: UpdateDeckUseCase = FromDI(UpdateDeckUseCase),
 ) -> schemas.Deck:
-    async with transaction:
-        instance = await decks_repository.update(data=data.model_dump(), item_id=deck_id)
-        await transaction.commit()
-    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/")
-@postgres_retry
 async def create_deck(
     data: schemas.DeckCreate,
-    decks_repository: DecksRepository = FromDI(DecksRepository),
-    transaction: Transaction = FromDI(Transaction),
+    create_deck_use_case: CreateDeckUseCase = FromDI(CreateDeckUseCase),
 ) -> schemas.Deck:
-    async with transaction:
-        instance = await decks_repository.create(data.model_dump())
-        await transaction.commit()
-    return schemas.Deck.model_validate(instance)
+    return schemas.Deck.model_validate(await create_deck_use_case(data=data))
diff --git a/app/application.py b/app/application.py
index 882bd52..4b45ea3 100644
--- a/app/application.py
+++ b/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()],
diff --git a/app/ioc.py b/app/ioc.py
index 6bc0e73..290e5a7 100644
--- a/app/ioc.py
+++ b/app/ioc.py
@@ -1,3 +1,5 @@
+import typing
+
 from db_retry import Transaction
 from modern_di import Group, Scope, providers
 
@@ -10,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),
@@ -40,13 +50,27 @@ class Dependencies(Group):
         kwargs={"session": session},
     )
 
+
+class Repositories(Group, scope=Scope.REQUEST):
     decks_repository = providers.Factory(
-        scope=Scope.REQUEST,
         creator=DecksRepository,
-        kwargs={"auto_commit": False, "session": session},
+        kwargs={"session": Database.session, "auto_commit": False},
     )
     cards_repository = providers.Factory(
-        scope=Scope.REQUEST,
         creator=CardsRepository,
-        kwargs={"auto_commit": False, "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]
diff --git a/app/use_cases/__init__.py b/app/use_cases/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/use_cases/create_cards.py b/app/use_cases/create_cards.py
new file mode 100644
index 0000000..cd491f6
--- /dev/null
+++ b/app/use_cases/create_cards.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
diff --git a/app/use_cases/create_deck.py b/app/use_cases/create_deck.py
new file mode 100644
index 0000000..80713d3
--- /dev/null
+++ b/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, 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
diff --git a/app/use_cases/get_card.py b/app/use_cases/get_card.py
new file mode 100644
index 0000000..650a8c8
--- /dev/null
+++ b/app/use_cases/get_card.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
diff --git a/app/use_cases/get_deck.py b/app/use_cases/get_deck.py
new file mode 100644
index 0000000..03e6837
--- /dev/null
+++ b/app/use_cases/get_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 DecksRepository
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class GetDeckUseCase:
+    transaction: Transaction
+    decks_repository: DecksRepository
+
+    @postgres_retry
+    async def __call__(self, *, deck_id: int) -> models.Deck:
+        async with self.transaction:
+            deck: typing.Final = await self.decks_repository.fetch_with_cards(deck_id)
+            await self.transaction.commit()
+        return deck
diff --git a/app/use_cases/list_cards.py b/app/use_cases/list_cards.py
new file mode 100644
index 0000000..5cda202
--- /dev/null
+++ b/app/use_cases/list_cards.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
+from app.repositories import CardsRepository
+
+
+@dataclasses.dataclass(kw_only=True, frozen=True, slots=True)
+class ListCardsUseCase:
+    transaction: Transaction
+    cards_repository: CardsRepository
+
+    @postgres_retry
+    async def __call__(self, *, deck_id: int) -> Sequence[models.Card]:
+        async with self.transaction:
+            cards: typing.Final = await self.cards_repository.list_for_deck(deck_id)
+            await self.transaction.commit()
+        return cards
diff --git a/app/use_cases/list_decks.py b/app/use_cases/list_decks.py
new file mode 100644
index 0000000..f1f2cce
--- /dev/null
+++ b/app/use_cases/list_decks.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
+from app.repositories import DecksRepository
+
+
+@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
diff --git a/app/use_cases/update_cards.py b/app/use_cases/update_cards.py
new file mode 100644
index 0000000..6a6774c
--- /dev/null
+++ b/app/use_cases/update_cards.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 UpdateCardsUseCase:
+    transaction: Transaction
+    cards_repository: CardsRepository
+
+    @postgres_retry
+    async def __call__(self, *, deck_id: int, data: list[schemas.Card]) -> Sequence[models.Card]:
+        async with self.transaction:
+            cards: typing.Final = await self.cards_repository.upsert_cards(deck_id, data)
+            await self.transaction.commit()
+        return cards
diff --git a/app/use_cases/update_deck.py b/app/use_cases/update_deck.py
new file mode 100644
index 0000000..303238a
--- /dev/null
+++ b/app/use_cases/update_deck.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 UpdateDeckUseCase:
+    transaction: Transaction
+    decks_repository: DecksRepository
+
+    @postgres_retry
+    async def __call__(self, *, deck_id: int, data: schemas.DeckCreate) -> models.Deck:
+        async with self.transaction:
+            deck: typing.Final = await self.decks_repository.update(data=data.model_dump(), item_id=deck_id)
+            await self.transaction.commit()
+        return deck
diff --git a/pyproject.toml b/pyproject.toml
index 0f9939a..440d757 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -70,6 +70,9 @@ isort.no-lines-before = ["standard-library", "local-folder"]
 "app/resources/*.py" = [
     "TC", # modern-di reads DI creator annotations at runtime
 ]
+"app/use_cases/*.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 325d75b..41f1c15 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -47,7 +47,7 @@ async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[
     engine = create_primary_sa_engine()
     connection = await engine.connect()
     transaction = await connection.begin()
-    di_container.override(ioc.Dependencies.dynamic_engine, connection)
+    di_container.override(ioc.Database.dynamic_engine, connection)
 
     try:
         yield create_session(connection)
diff --git a/tests/test_connection_loss.py b/tests/test_connection_loss.py
index aebe4c3..9326d68 100644
--- a/tests/test_connection_loss.py
+++ b/tests/test_connection_loss.py
@@ -1,5 +1,4 @@
 import asyncio
-import inspect
 import typing
 
 import asyncpg
@@ -10,7 +9,6 @@
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession
 
 from app import ioc, models, schemas
-from app.api import cards, decks
 from app.resources.db import create_primary_sa_engine, create_read_sa_engine
 
 
@@ -52,8 +50,8 @@ async def stored_ids() -> typing.AsyncIterator[StoredIds]:
 async def engines(di_container: modern_di.Container) -> typing.AsyncIterator[tuple[AsyncEngine, AsyncEngine]]:
     primary_engine: typing.Final = create_primary_sa_engine()
     read_engine: typing.Final = create_read_sa_engine()
-    di_container.override(ioc.Dependencies.database_engine, primary_engine)
-    di_container.override(ioc.Dependencies.database_read_engine, read_engine)
+    di_container.override(ioc.Database.database_engine, primary_engine)
+    di_container.override(ioc.Database.database_read_engine, read_engine)
     try:
         yield primary_engine, read_engine
     finally:
@@ -91,7 +89,7 @@ async def _terminate_backend(engine: AsyncEngine, pid: int, closed: asyncio.Even
 @pytest.fixture
 async def lost_connection(engines: tuple[AsyncEngine, AsyncEngine], request_container: modern_di.Container) -> None:
     """Leave the request session holding a primary connection whose backend was terminated."""
-    session: typing.Final = request_container.resolve_provider(ioc.Dependencies.session)
+    session: typing.Final = request_container.resolve_provider(ioc.Database.session)
     assert isinstance(session, AsyncSession)
     connection: typing.Final = await session.connection()
     primary_engine, _ = engines
@@ -133,32 +131,32 @@ async def test_endpoint_recovers_from_a_connection_lost_while_pooled(
 
 @pytest.mark.usefixtures("lost_connection")
 @pytest.mark.parametrize(
-    ("handler", "build_kwargs"),
+    ("provider", "build_kwargs"),
     [
-        (decks.list_decks, lambda _: {}),
-        (decks.get_deck, lambda ids: {"deck_id": ids.deck_id}),
-        (decks.update_deck, lambda ids: {"deck_id": ids.deck_id, "data": schemas.DeckCreate(name="renamed")}),
-        (decks.create_deck, lambda _: {"data": schemas.DeckCreate(name="new")}),
-        (cards.list_cards, lambda ids: {"deck_id": ids.deck_id}),
-        (cards.get_card, lambda ids: {"card_id": ids.card_id}),
-        (cards.create_cards, lambda ids: {"deck_id": ids.deck_id, "data": [schemas.CardCreate(front="new")]}),
+        (ioc.UseCases.list_decks_use_case, lambda _: {}),
+        (ioc.UseCases.get_deck_use_case, lambda ids: {"deck_id": ids.deck_id}),
         (
-            cards.update_cards,
+            ioc.UseCases.update_deck_use_case,
+            lambda ids: {"deck_id": ids.deck_id, "data": schemas.DeckCreate(name="renamed")},
+        ),
+        (ioc.UseCases.create_deck_use_case, lambda _: {"data": schemas.DeckCreate(name="new")}),
+        (ioc.UseCases.list_cards_use_case, lambda ids: {"deck_id": ids.deck_id}),
+        (ioc.UseCases.get_card_use_case, lambda ids: {"card_id": ids.card_id}),
+        (
+            ioc.UseCases.create_cards_use_case,
+            lambda ids: {"deck_id": ids.deck_id, "data": [schemas.CardCreate(front="new")]},
+        ),
+        (
+            ioc.UseCases.update_cards_use_case,
             lambda ids: {"deck_id": ids.deck_id, "data": [schemas.Card(id=ids.card_id, front="renamed")]},
         ),
     ],
 )
-async def test_handler_recovers_from_a_lost_connection(
+async def test_use_case_recovers_from_a_lost_connection(
     request_container: modern_di.Container,
     stored_ids: StoredIds,
-    handler: typing.Callable[..., typing.Awaitable[schemas.Base]],
+    provider: modern_di.providers.Factory[typing.Any],
     build_kwargs: typing.Callable[[StoredIds], dict[str, typing.Any]],
 ) -> None:
-    dependencies: typing.Final = {
-        "decks_repository": request_container.resolve_provider(ioc.Dependencies.decks_repository),
-        "cards_repository": request_container.resolve_provider(ioc.Dependencies.cards_repository),
-        "transaction": request_container.resolve_provider(ioc.Dependencies.transaction),
-    }
-    parameters: typing.Final = inspect.signature(handler).parameters
-    kwargs: typing.Final = {name: value for name, value in dependencies.items() if name in parameters}
-    assert isinstance(await handler(**kwargs, **build_kwargs(stored_ids)), schemas.Base)
+    use_case: typing.Final = request_container.resolve_provider(provider)
+    assert await use_case(**build_kwargs(stored_ids))
diff --git a/tests/test_db_routing.py b/tests/test_db_routing.py
index a2da494..e0a56e6 100644
--- a/tests/test_db_routing.py
+++ b/tests/test_db_routing.py
@@ -19,7 +19,7 @@
 @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)
+    di_container.override(ioc.Database.database_engine, engine)
     yield engine
     await engine.dispose()
 
@@ -27,7 +27,7 @@ async def primary_engine(di_container: modern_di.Container) -> typing.AsyncItera
 @pytest.fixture
 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)
+    di_container.override(ioc.Database.database_read_engine, engine)
     yield engine
     await engine.dispose()
 
@@ -35,7 +35,7 @@ async def read_engine(di_container: modern_di.Container) -> typing.AsyncIterator
 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)
+        return request_container.resolve_provider(ioc.Database.dynamic_engine)
 
 
 @pytest.mark.parametrize("method", ["GET", "HEAD"])
diff --git a/tests/test_main.py b/tests/test_main.py
index 1050a3a..1a9ecd3 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -18,7 +18,7 @@ def test_main(monkeypatch: pytest.MonkeyPatch) -> None:
 
 
 async def test_session() -> None:
-    container = modern_di.Container(groups=[ioc.Dependencies])
+    container = modern_di.Container(groups=ioc.ALL_GROUPS)
     request_container = container.build_child_container(scope=modern_di.Scope.REQUEST)
     try:
         request_container.resolve(AsyncSession)

From 83c870ca7ee8528bf5c6fde0a443f53fc1f86072 Mon Sep 17 00:00:00 2001
From: Artur Shiriev <lesnik512@gmail.com>
Date: Fri, 16 Oct 2026 22:00:23 +0300
Subject: [PATCH 4/5] 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.
---
 AGENTS.md                     |   2 +-
 tests/test_connection_loss.py | 140 +++++-----------------------------
 2 files changed, 22 insertions(+), 120 deletions(-)

diff --git a/AGENTS.md b/AGENTS.md
index 3a71615..3ca1998 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -96,7 +96,7 @@ class ListDecksUseCase:
 - `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")]`.
 
-`tests/test_connection_loss.py` overrides both engines with fresh ones and terminates a backend with `pg_terminate_backend`: `lost_pooled_connection` kills an idle pooled connection on each engine and goes through the HTTP client, `lost_connection` kills the connection a request session already holds and calls each use case directly. These tests commit for real, so `stored_ids` truncates `decks` and `cards` before and after. Each retry sleeps 1 to 2 seconds, which is most of the suite's run time.
+`tests/test_connection_loss.py` overrides the primary engine with a fresh one and terminates a backend with `pg_terminate_backend`. The `lost_connection` fixture runs twice: once it kills the connection the request session already holds, once an idle connection in the pool. `ListDecksUseCase` then has to recover through the retry. Every use case has the same shape, so one use case stands in for all of them. Run with `DB_RETRY_RETRIES_NUMBER=1` and the test fails.
 
 `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.
 
diff --git a/tests/test_connection_loss.py b/tests/test_connection_loss.py
index 9326d68..d8208f3 100644
--- a/tests/test_connection_loss.py
+++ b/tests/test_connection_loss.py
@@ -5,65 +5,22 @@
 import modern_di
 import pytest
 import sqlalchemy as sa
-from fastapi import status
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession
 
-from app import ioc, models, schemas
-from app.resources.db import create_primary_sa_engine, create_read_sa_engine
-
-
-if typing.TYPE_CHECKING:
-    from httpx import AsyncClient
-
-
-class StoredIds(typing.NamedTuple):
-    deck_id: int
-    card_id: int
-
-
-async def _truncate(engine: AsyncEngine) -> None:
-    async with engine.begin() as connection:
-        await connection.execute(sa.text("TRUNCATE cards, decks RESTART IDENTITY"))
+from app import ioc
+from app.resources.db import create_primary_sa_engine
 
 
 @pytest.fixture
-async def stored_ids() -> typing.AsyncIterator[StoredIds]:
+async def request_container(di_container: modern_di.Container) -> typing.AsyncIterator[modern_di.Container]:
     engine: typing.Final = create_primary_sa_engine()
-    await _truncate(engine)
-    async with engine.begin() as connection:
-        deck_id: typing.Final = (
-            await connection.execute(sa.insert(models.Deck).values(name="deck").returning(models.Deck.id))
-        ).scalar_one()
-        card_id: typing.Final = (
-            await connection.execute(
-                sa.insert(models.Card).values(front="front", deck_id=deck_id).returning(models.Card.id)
-            )
-        ).scalar_one()
-    try:
-        yield StoredIds(deck_id=deck_id, card_id=card_id)
-    finally:
-        await _truncate(engine)
-        await engine.dispose()
-
-
-@pytest.fixture
-async def engines(di_container: modern_di.Container) -> typing.AsyncIterator[tuple[AsyncEngine, AsyncEngine]]:
-    primary_engine: typing.Final = create_primary_sa_engine()
-    read_engine: typing.Final = create_read_sa_engine()
-    di_container.override(ioc.Database.database_engine, primary_engine)
-    di_container.override(ioc.Database.database_read_engine, read_engine)
+    di_container.override(ioc.Database.database_engine, engine)
     try:
-        yield primary_engine, read_engine
+        async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
+            yield container
     finally:
         di_container.reset_override()
-        await primary_engine.dispose()
-        await read_engine.dispose()
-
-
-@pytest.fixture
-async def request_container(di_container: modern_di.Container) -> typing.AsyncIterator[modern_di.Container]:
-    async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
-        yield container
+        await engine.dispose()
 
 
 async def _watch_backend(connection: AsyncConnection) -> tuple[int, asyncio.Event]:
@@ -86,77 +43,22 @@ async def _terminate_backend(engine: AsyncEngine, pid: int, closed: asyncio.Even
     await asyncio.wait_for(closed.wait(), timeout=5)
 
 
-@pytest.fixture
-async def lost_connection(engines: tuple[AsyncEngine, AsyncEngine], request_container: modern_di.Container) -> None:
-    """Leave the request session holding a primary connection whose backend was terminated."""
-    session: typing.Final = request_container.resolve_provider(ioc.Database.session)
-    assert isinstance(session, AsyncSession)
-    connection: typing.Final = await session.connection()
-    primary_engine, _ = engines
-    await _terminate_backend(primary_engine, *await _watch_backend(connection))
-
-
-@pytest.fixture
-async def lost_pooled_connection(engines: tuple[AsyncEngine, AsyncEngine]) -> None:
-    """Leave each engine's pool holding one idle connection whose backend was terminated."""
-    for engine in engines:
+@pytest.fixture(params=["held_by_session", "idle_in_pool"])
+async def lost_connection(request: pytest.FixtureRequest, request_container: modern_di.Container) -> None:
+    """Terminate the backend of the connection the request session holds, or of an idle pooled one."""
+    engine: typing.Final = request_container.resolve_provider(ioc.Database.database_engine)
+    assert isinstance(engine, AsyncEngine)
+    if request.param == "held_by_session":
+        session: typing.Final = request_container.resolve_provider(ioc.Database.session)
+        assert isinstance(session, AsyncSession)
+        watched = await _watch_backend(await session.connection())
+    else:
         async with engine.connect() as connection:
             watched = await _watch_backend(connection)
-        await _terminate_backend(engine, *watched)
-
-
-@pytest.mark.usefixtures("lost_pooled_connection")
-@pytest.mark.parametrize(
-    "build_request",
-    [
-        lambda _: ("GET", "/api/decks/", None),
-        lambda ids: ("GET", f"/api/decks/{ids.deck_id}/", None),
-        lambda ids: ("PUT", f"/api/decks/{ids.deck_id}/", {"name": "renamed"}),
-        lambda _: ("POST", "/api/decks/", {"name": "new"}),
-        lambda ids: ("GET", f"/api/decks/{ids.deck_id}/cards/", None),
-        lambda ids: ("GET", f"/api/cards/{ids.card_id}/", None),
-        lambda ids: ("POST", f"/api/decks/{ids.deck_id}/cards/", [{"front": "new"}]),
-        lambda ids: ("PUT", f"/api/decks/{ids.deck_id}/cards/", [{"id": ids.card_id, "front": "renamed"}]),
-    ],
-)
-async def test_endpoint_recovers_from_a_connection_lost_while_pooled(
-    client: AsyncClient,
-    stored_ids: StoredIds,
-    build_request: typing.Callable[[StoredIds], tuple[str, str, object]],
-) -> None:
-    method, path, json = build_request(stored_ids)
-    response: typing.Final = await client.request(method, path, json=json)
-    assert response.status_code == status.HTTP_200_OK
+    await _terminate_backend(engine, *watched)
 
 
 @pytest.mark.usefixtures("lost_connection")
-@pytest.mark.parametrize(
-    ("provider", "build_kwargs"),
-    [
-        (ioc.UseCases.list_decks_use_case, lambda _: {}),
-        (ioc.UseCases.get_deck_use_case, lambda ids: {"deck_id": ids.deck_id}),
-        (
-            ioc.UseCases.update_deck_use_case,
-            lambda ids: {"deck_id": ids.deck_id, "data": schemas.DeckCreate(name="renamed")},
-        ),
-        (ioc.UseCases.create_deck_use_case, lambda _: {"data": schemas.DeckCreate(name="new")}),
-        (ioc.UseCases.list_cards_use_case, lambda ids: {"deck_id": ids.deck_id}),
-        (ioc.UseCases.get_card_use_case, lambda ids: {"card_id": ids.card_id}),
-        (
-            ioc.UseCases.create_cards_use_case,
-            lambda ids: {"deck_id": ids.deck_id, "data": [schemas.CardCreate(front="new")]},
-        ),
-        (
-            ioc.UseCases.update_cards_use_case,
-            lambda ids: {"deck_id": ids.deck_id, "data": [schemas.Card(id=ids.card_id, front="renamed")]},
-        ),
-    ],
-)
-async def test_use_case_recovers_from_a_lost_connection(
-    request_container: modern_di.Container,
-    stored_ids: StoredIds,
-    provider: modern_di.providers.Factory[typing.Any],
-    build_kwargs: typing.Callable[[StoredIds], dict[str, typing.Any]],
-) -> None:
-    use_case: typing.Final = request_container.resolve_provider(provider)
-    assert await use_case(**build_kwargs(stored_ids))
+async def test_use_case_recovers_from_a_lost_connection(request_container: modern_di.Container) -> None:
+    use_case: typing.Final = request_container.resolve_provider(ioc.UseCases.list_decks_use_case)
+    assert await use_case() == []

From 0274b3f34e38b75a2ea9e54e60c0493e25961036 Mon Sep 17 00:00:00 2001
From: Artur Shiriev <lesnik512@gmail.com>
Date: Fri, 16 Oct 2026 22:05:31 +0300
Subject: [PATCH 5/5] 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.
---
 AGENTS.md                     |  2 --
 tests/test_connection_loss.py | 64 -----------------------------------
 2 files changed, 66 deletions(-)
 delete mode 100644 tests/test_connection_loss.py

diff --git a/AGENTS.md b/AGENTS.md
index 3ca1998..8c5fb9d 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -96,8 +96,6 @@ class ListDecksUseCase:
 - `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")]`.
 
-`tests/test_connection_loss.py` overrides the primary engine with a fresh one and terminates a backend with `pg_terminate_backend`. The `lost_connection` fixture runs twice: once it kills the connection the request session already holds, once an idle connection in the pool. `ListDecksUseCase` then has to recover through the retry. Every use case has the same shape, so one use case stands in for all of them. Run with `DB_RETRY_RETRIES_NUMBER=1` and the test fails.
-
 `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.
 
 ## Conventions
diff --git a/tests/test_connection_loss.py b/tests/test_connection_loss.py
deleted file mode 100644
index d8208f3..0000000
--- a/tests/test_connection_loss.py
+++ /dev/null
@@ -1,64 +0,0 @@
-import asyncio
-import typing
-
-import asyncpg
-import modern_di
-import pytest
-import sqlalchemy as sa
-from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine, AsyncSession
-
-from app import ioc
-from app.resources.db import create_primary_sa_engine
-
-
-@pytest.fixture
-async def request_container(di_container: modern_di.Container) -> typing.AsyncIterator[modern_di.Container]:
-    engine: typing.Final = create_primary_sa_engine()
-    di_container.override(ioc.Database.database_engine, engine)
-    try:
-        async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
-            yield container
-    finally:
-        di_container.reset_override()
-        await engine.dispose()
-
-
-async def _watch_backend(connection: AsyncConnection) -> tuple[int, asyncio.Event]:
-    pid: typing.Final = await connection.scalar(sa.text("SELECT pg_backend_pid()"))
-    driver_connection: typing.Final = (await connection.get_raw_connection()).driver_connection
-    assert driver_connection is not None
-    closed: typing.Final = asyncio.Event()
-    driver_connection.add_termination_listener(lambda _: closed.set())
-    return pid, closed
-
-
-async def _terminate_backend(engine: AsyncEngine, pid: int, closed: asyncio.Event) -> None:
-    killer: typing.Final = await asyncpg.connect(
-        engine.url.set(drivername="postgresql").render_as_string(hide_password=False)
-    )
-    try:
-        await killer.execute("SELECT pg_terminate_backend($1)", pid)
-    finally:
-        await killer.close()
-    await asyncio.wait_for(closed.wait(), timeout=5)
-
-
-@pytest.fixture(params=["held_by_session", "idle_in_pool"])
-async def lost_connection(request: pytest.FixtureRequest, request_container: modern_di.Container) -> None:
-    """Terminate the backend of the connection the request session holds, or of an idle pooled one."""
-    engine: typing.Final = request_container.resolve_provider(ioc.Database.database_engine)
-    assert isinstance(engine, AsyncEngine)
-    if request.param == "held_by_session":
-        session: typing.Final = request_container.resolve_provider(ioc.Database.session)
-        assert isinstance(session, AsyncSession)
-        watched = await _watch_backend(await session.connection())
-    else:
-        async with engine.connect() as connection:
-            watched = await _watch_backend(connection)
-    await _terminate_backend(engine, *watched)
-
-
-@pytest.mark.usefixtures("lost_connection")
-async def test_use_case_recovers_from_a_lost_connection(request_container: modern_di.Container) -> None:
-    use_case: typing.Final = request_container.resolve_provider(ioc.UseCases.list_decks_use_case)
-    assert await use_case() == []