lesnik512 · Oct 17, 2026 · Oct 17, 2026
diff --git a/AGENTS.md b/AGENTS.md
@@ -43,7 +43,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
 
 `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.
+- `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 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` — `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.
@@ -84,7 +84,7 @@ class ListDecksUseCase:
 ### 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])`. Use cases call the service methods (`get_many`, `get_one`, `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_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`.
 - `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`.
 
@@ -106,7 +106,7 @@ class ListDecksUseCase:
 
 - Ruff is configured with `select = ["ALL"]` and a curated ignore list in `pyproject.toml`. Don't sprinkle `# noqa`; prefer fixing or extending the project ignore list if a rule is genuinely wrong for the codebase.
 - Routes convert ORM objects to schemas explicitly, never via `typing.cast`. Single objects: `schemas.X.model_validate(instance)`. Collections: `schemas.Xs.from_models(objects)` (the `Collection[T]` seam in `app/schemas.py`). Both rely on `from_attributes=True`.
-- 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/writes 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/writes use `noload` (no cards query), detail uses `joinedload` via `fetch_with_cards` — so the type states exactly what each endpoint loads.
 - Line length is 120.
 
 ## Agent skills

diff --git a/app/repositories.py b/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"))
diff --git a/app/use_cases/update_deck.py b/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
diff --git a/tests/conftest.py b/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 app import ioc
 from app.application import build_app
@@ -14,9 +15,13 @@
 if typing.TYPE_CHECKING:
     import fastapi
     import modern_di
+    from sqlalchemy.engine import Connection
     from sqlalchemy.ext.asyncio import AsyncSession
 
 
+SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
+
+
 @pytest.fixture
 async def app() -> typing.AsyncIterator[fastapi.FastAPI]:
     app_ = build_app()
@@ -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)
diff --git a/tests/test_cards.py b/tests/test_cards.py
@@ -1,13 +1,17 @@
+import datetime
 from typing import TYPE_CHECKING
 
 import pytest
+import sqlalchemy as sa
 from fastapi import status
 
+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")]
@@ -137,6 +141,47 @@ async def test_cards_ordered_by_id_after_update(client: AsyncClient) -> None:
     assert [x["id"] for x in response.json()["items"]] == [card1.id, card2.id]
 
 
+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.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.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.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.HTTP_422_UNPROCESSABLE_CONTENT
+
+
 async def test_openapi_card_deck_id_required(client: AsyncClient) -> None:
     response = await client.get("/openapi.json")
     assert response.status_code == status.HTTP_200_OK

diff --git a/tests/test_decks.py b/tests/test_decks.py
@@ -1,14 +1,18 @@
+import datetime
 import uuid
 from typing import TYPE_CHECKING
 
 import pytest
+import sqlalchemy as sa
 from fastapi import status
 
+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.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.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.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