Cut round trips on deck detail, deck update and card upsert (#72)

Deck detail loads cards with joinedload (one query instead of two). Deck
update is one UPDATE ... RETURNING instead of advanced-alchemy's update,
which read the row twice and refreshed it. Card upsert is one INSERT ...
ON CONFLICT (id) DO UPDATE ... RETURNING instead of upsert_many, which
selected the rows before and after writing them.

Author: lesnik512

AGENTS.md

@@ -26,7 +26,7 @@ Python is 3.14, dependencies managed by `uv`. The API is exposed on `:8000`.
 
 `Database` group:
 - `database_engine` — application-scoped primary engine, finalizer disposes the engine
-- `database_read_engine` — application-scoped read engine 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
+- `database_read_engine` — application-scoped read engine 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 read that runs several queries can see a write that commits in between. Load related rows in the same query (`fetch_with_cards` uses `joinedload`). 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` — request-scoped; `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` — request-scoped, bound to `dynamic_engine`, finalizer closes the session
 - `transaction` — request-scoped `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session
@@ -58,7 +58,7 @@ class ListDecksUseCase:
   - `CreateCardsUseCase` relies on the `(deck_id, front)` unique constraint. A replay is rejected as a duplicate-key error rather than creating the cards twice.
   - Updates (`UpdateDeckUseCase`, `UpdateCardsUseCase`) write the same values again, so a replay is harmless.
 
-**Persistence**: Models inherit `advanced_alchemy.base.BigIntAuditBase` (gives `id: BigInt`, `created_at`, `updated_at`). Metadata is shared with `orm.DeclarativeBase.metadata` in `app/models.py` so Alembic autogen sees everything. Repositories are `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. `create_session` in `app/resources/db.py` builds a plain `AsyncSession` with `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the per-test rollback fixture: 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.
+**Persistence**: Models inherit `advanced_alchemy.base.BigIntAuditBase` (gives `id: BigInt`, `created_at`, `updated_at`). Metadata is shared with `orm.DeclarativeBase.metadata` in `app/models.py` so Alembic autogen sees everything. Repositories are `SQLAlchemyAsyncRepositoryService[Model]` with a nested `BaseRepository(SQLAlchemyAsyncRepository[Model])`. Use cases call the service methods (`get_many`, `get_one`, `create_many`) for reads and plain inserts. Writes that advanced-alchemy would spread over several round trips are one statement on the session instead: its `update` reads the row twice and refreshes it after the flush, and `upsert_many` selects the rows before and after writing them. `update_deck` sends `UPDATE ... RETURNING`, and `get_or_create` and `upsert_cards` send `INSERT ... ON CONFLICT ... RETURNING`. A statement that can hit a unique constraint runs inside `wrap_sqlalchemy_exception` so that it still raises `DuplicateKeyError`. `create_session` in `app/resources/db.py` builds a plain `AsyncSession` with `join_transaction_mode="create_savepoint"`. This is inert in production (the session binds to an engine) but enables the per-test rollback fixture: 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.
 
 **Test isolation** (`tests/conftest.py`): `db_session` fixture opens a connection, starts a transaction, then **overrides** `Database.dynamic_engine` in the DI container to return that connection. All requests in the test reuse this connection. Each session built against it uses `join_transaction_mode="create_savepoint"`, so `transaction.commit()` releases the session's own savepoint while the outer transaction is rolled back in teardown. DB state is clean between tests with no truncation needed. `app` and `client` fixtures build the real app and run it through `httpx.ASGITransport` + `asgi_lifespan.LifespanManager`. Polyfactory `SQLAlchemyFactory` is wired up via `set_async_session_in_base_sqlalchemy_factory`.
 
@@ -70,7 +70,7 @@ class ListDecksUseCase:
 
 - Routes live in `app/api/`, one module per resource (`decks.py`, `cards.py`), each exposing its own `ROUTER` (`litestar.Router`, prefix `/api`). `application.build_app` registers them all via `route_handlers=[decks.ROUTER, cards.ROUTER]`. Add a new resource by creating `app/api/<name>.py`, defining handlers + a `ROUTER`, and adding it to that list. Its database work goes in `app/use_cases/` with a provider in `ioc.UseCases`.
 - Pydantic schemas in `app/schemas.py` use `from_attributes=True` (via `Base`) so they validate directly from ORM instances (`schemas.X.model_validate(orm_instance)`). Collection responses go through `Collection[T].from_models(...)` (e.g. `schemas.Decks`, `schemas.Cards`).
-- Deck responses are deliberately two-shaped: `list_decks`/`create_deck`/`update_deck` return the light `schemas.Deck` (no `cards`), while `get_deck` returns `schemas.DeckWithCards`. The split mirrors loading — lists use `noload` (no cards query), detail uses `selectinload` via `fetch_with_cards` — so the type states exactly what each endpoint loads.
+- Deck responses are deliberately two-shaped: `list_decks`/`create_deck`/`update_deck` return the light `schemas.Deck` (no `cards`), while `get_deck` returns `schemas.DeckWithCards`. The split mirrors loading — lists use `noload` (no cards query), detail uses `joinedload` via `fetch_with_cards` — so the type states exactly what each endpoint loads.
 - Domain exceptions: register handlers in `application.build_app`'s `exception_handlers` dict (`DuplicateKeyError` → `exceptions.duplicate_key_error_handler` returns 400, `NotFoundError` → `exceptions.not_found_error_handler` returns 404). Handlers don't raise HTTP errors themselves: a missing row surfaces as advanced-alchemy's `NotFoundError`.
 - `ruff` is configured with `select = ["ALL"]` and a line length of 120 — expect strict lint. `app/resources/` is exempt from the `TC` rules because modern-di resolves DI creator annotations at runtime. Type-check with `ty`; suppressions already exist for `invalid-argument-type` around `LifespanManager` / `ASGITransport` in `tests/conftest.py`.
 

app/repositories.py

@@ -1,7 +1,10 @@
 import operator
+import typing
 from typing import TYPE_CHECKING
 
-from advanced_alchemy.repository import SQLAlchemyAsyncRepository
+import sqlalchemy as sa
+from advanced_alchemy.exceptions import NotFoundError, wrap_sqlalchemy_exception
+from advanced_alchemy.repository import DEFAULT_ERROR_MESSAGE_TEMPLATES, SQLAlchemyAsyncRepository
 from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
 from sqlalchemy import orm
 from sqlalchemy.dialects import postgresql
@@ -33,9 +36,19 @@ async def get_or_create(self, data: schemas.DeckCreate) -> models.Deck:
     async def fetch_with_cards(self, deck_id: int) -> models.Deck:
         return await self.get_one(
             models.Deck.id == deck_id,
-            load=[orm.selectinload(models.Deck.cards)],
+            load=[orm.joinedload(models.Deck.cards)],
+            uniquify=True,
         )
 
+    async def update_deck(self, deck_id: int, data: schemas.DeckUpdate) -> models.Deck:
+        statement: typing.Final = (
+            sa.update(models.Deck).where(models.Deck.id == deck_id).values(**data.model_dump()).returning(models.Deck)
+        )
+        deck: typing.Final = (await self.repository.session.scalars(statement)).one_or_none()
+        if deck is None:
+            raise NotFoundError
+        return deck
+
 
 class CardsRepository(SQLAlchemyAsyncRepositoryService[models.Card]):
     class BaseRepository(SQLAlchemyAsyncRepository[models.Card]):
@@ -50,7 +63,19 @@ async def add_cards(self, deck_id: int, cards: list[schemas.CardCreate]) -> Sequ
         return await self.create_many([models.Card(**card.model_dump(), deck_id=deck_id) for card in cards])
 
     async def upsert_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> Sequence[models.Card]:
-        upserted = await self.upsert_many(
-            [models.Card(**card.model_dump(), deck_id=deck_id) for card in cards],
+        if not cards:
+            return []
+        statement = postgresql.insert(models.Card).values([{**card.model_dump(), "deck_id": deck_id} for card in cards])
+        statement = statement.on_conflict_do_update(
+            index_elements=[models.Card.id],
+            set_={
+                "front": statement.excluded.front,
+                "back": statement.excluded.back,
+                "hint": statement.excluded.hint,
+                "deck_id": statement.excluded.deck_id,
+                "updated_at": sa.func.now(),
+            },
         )
+        with wrap_sqlalchemy_exception(error_messages=DEFAULT_ERROR_MESSAGE_TEMPLATES, dialect_name="postgresql"):
+            upserted: typing.Final = (await self.repository.session.scalars(statement.returning(models.Card))).all()
         return sorted(upserted, key=operator.attrgetter("id"))

app/use_cases/update_deck.py

@@ -15,6 +15,6 @@ class UpdateDeckUseCase:
     @postgres_retry
     async def __call__(self, *, deck_id: int, data: schemas.DeckUpdate) -> models.Deck:
         async with self.transaction:
-            deck: typing.Final = await self.decks_repository.update(data=data.model_dump(), item_id=deck_id)
+            deck: typing.Final = await self.decks_repository.update_deck(deck_id, data)
             await self.transaction.commit()
         return deck

tests/conftest.py

@@ -5,6 +5,7 @@
 from asgi_lifespan import LifespanManager
 from httpx import ASGITransport, AsyncClient
 from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
+from sqlalchemy import event
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app import ioc
@@ -15,6 +16,10 @@
 if typing.TYPE_CHECKING:
     import litestar
     import modern_di
+    from sqlalchemy.engine import Connection
+
+
+SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
 
 
 @pytest.fixture
@@ -68,3 +73,20 @@ async def set_async_session_in_base_sqlalchemy_factory(
         yield
     finally:
         SQLAlchemyFactory.__async_session__ = None
+
+
+@pytest.fixture
+def statements(db_session: AsyncSession) -> typing.Iterator[list[str]]:
+    sync_connection: typing.Final = typing.cast("Connection", db_session.sync_session.bind)
+    recorded: typing.Final[list[str]] = []
+
+    def record(*args: object) -> None:
+        statement: typing.Final = typing.cast("str", args[2])
+        if not statement.startswith(SAVEPOINT_PREFIXES):
+            recorded.append(statement)
+
+    event.listen(sync_connection, "before_cursor_execute", record)
+    try:
+        yield recorded
+    finally:
+        event.remove(sync_connection, "before_cursor_execute", record)

tests/test_cards.py

@@ -1,13 +1,17 @@
+import datetime
 from typing import TYPE_CHECKING
 
 import pytest
+import sqlalchemy as sa
 from litestar import status_codes
 
+from app import models
 from tests import factories
 
 
 if TYPE_CHECKING:
     from httpx import AsyncClient
+    from sqlalchemy.ext.asyncio import AsyncSession
 
 
 pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")]
@@ -143,3 +147,44 @@ async def test_openapi_card_deck_id_required(client: AsyncClient) -> None:
     card_schema = response.json()["components"]["schemas"]["Card"]
     assert "deck_id" in card_schema["required"]
     assert card_schema["properties"]["deck_id"]["type"] == "integer"
+
+
+async def test_update_cards_sends_one_statement(client: AsyncClient, statements: list[str]) -> None:
+    deck = await factories.DeckModelFactory.create_async()
+    card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
+    statements.clear()
+
+    response = await client.put(
+        f"/api/decks/{deck.id}/cards/",
+        json=[{"id": card1.id, "front": "card1 updated"}, {"id": card2.id, "front": "card2 updated"}],
+    )
+    assert response.status_code == status_codes.HTTP_200_OK
+    assert [x["front"] for x in response.json()["items"]] == ["card1 updated", "card2 updated"]
+    assert len(statements) == 1
+
+
+async def test_update_cards_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None:
+    long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
+    deck = await factories.DeckModelFactory.create_async()
+    card = await factories.CardModelFactory.create_async(deck_id=deck.id, updated_at=long_ago)
+
+    response = await client.put(f"/api/decks/{deck.id}/cards/", json=[{"id": card.id, "front": "updated"}])
+    assert response.status_code == status_codes.HTTP_200_OK
+    updated_at = (await db_session.scalars(sa.select(models.Card.updated_at).where(models.Card.id == card.id))).one()
+    assert updated_at > long_ago
+
+
+async def test_update_cards_with_no_cards(client: AsyncClient) -> None:
+    deck = await factories.DeckModelFactory.create_async()
+
+    response = await client.put(f"/api/decks/{deck.id}/cards/", json=[])
+    assert response.status_code == status_codes.HTTP_200_OK
+    assert response.json() == {"items": []}
+
+
+async def test_update_cards_to_a_duplicate_front(client: AsyncClient) -> None:
+    deck = await factories.DeckModelFactory.create_async()
+    card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
+
+    response = await client.put(f"/api/decks/{deck.id}/cards/", json=[{"id": card2.id, "front": card1.front}])
+    assert response.status_code == status_codes.HTTP_400_BAD_REQUEST

tests/test_decks.py

@@ -1,14 +1,18 @@
+import datetime
 import uuid
 from typing import TYPE_CHECKING
 
 import pytest
+import sqlalchemy as sa
 from litestar import status_codes
 
+from app import models
 from tests import factories
 
 
 if TYPE_CHECKING:
     from httpx import AsyncClient
+    from sqlalchemy.ext.asyncio import AsyncSession
 
 
 pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")]
@@ -72,6 +76,17 @@ async def test_get_one_deck(client: AsyncClient) -> None:
         assert v == getattr(deck, k)
 
 
+async def test_get_one_deck_loads_cards_in_one_query(client: AsyncClient, statements: list[str]) -> None:
+    deck = await factories.DeckModelFactory.create_async()
+    await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
+    statements.clear()
+
+    response = await client.get(f"/api/decks/{deck.id}/")
+    assert response.status_code == status_codes.HTTP_200_OK
+    assert len(response.json()["cards"]) == 2  # noqa: PLR2004
+    assert len(statements) == 1
+
+
 @pytest.mark.parametrize(
     ("name", "description", "status_code"),
     [
@@ -169,3 +184,23 @@ async def test_post_decks_twice_with_one_idempotency_key_creates_one_deck(client
     assert second.json() == first.json()
     response = await client.get("/api/decks/")
     assert [item["name"] for item in response.json()["items"]] == ["first"]
+
+
+async def test_put_decks_sends_one_statement(client: AsyncClient, statements: list[str]) -> None:
+    deck = await factories.DeckModelFactory.create_async()
+    statements.clear()
+
+    response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None})
+    assert response.status_code == status_codes.HTTP_200_OK
+    assert response.json() == {"id": deck.id, "name": "renamed", "description": None}
+    assert len(statements) == 1
+
+
+async def test_put_decks_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None:
+    long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
+    deck = await factories.DeckModelFactory.create_async(updated_at=long_ago)
+
+    response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None})
+    assert response.status_code == status_codes.HTTP_200_OK
+    updated_at = (await db_session.scalars(sa.select(models.Deck.updated_at).where(models.Deck.id == deck.id))).one()
+    assert updated_at > long_ago