Bound card writes and ids, map missing decks to 404, cache the card update

AGENTS.md

@@ -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])`. 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` sends `INSERT ... ON CONFLICT ... RETURNING`. `update_cards` sends one `UPDATE cards ... FROM (VALUES ...) WHERE cards.id = updates.id AND cards.deck_id = :deck_id RETURNING *` and raises `NotFoundError` (404) if fewer rows come back than ids were sent, which rolls the whole request back. PUT never inserts. An `INSERT ... ON CONFLICT (id)` upsert would move a card in from another deck and would insert client-chosen ids without advancing `cards_id_seq`, so a later POST would collide on the primary key. The `update_cards` handler rejects duplicate ids with a 400 before any query runs. 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.
+**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` sends `INSERT ... ON CONFLICT ... RETURNING`. `update_cards` sends one `UPDATE cards ... FROM unnest(:ids, :fronts, :backs, :hints) AS updates(...) WHERE cards.id = updates.id AND cards.deck_id = :deck_id RETURNING *`. It binds one array per column, so the SQL text stays the same for any number of cards: SQLAlchemy caches it and asyncpg prepares it once per connection, which a multi-row `VALUES` would not allow. It raises `NotFoundError` (404) if fewer rows come back than ids were sent, which rolls the whole request back. PUT never inserts. An `INSERT ... ON CONFLICT (id)` upsert would move a card in from another deck and would insert client-chosen ids without advancing `cards_id_seq`, so a later POST would collide on the primary key. The card handlers cap a request at `schemas.MAX_CARDS_PER_REQUEST` cards, and `update_cards` rejects duplicate ids, both with a 400 before any query runs. Ids in paths, `after` and card bodies are bounded to Postgres `bigint` (`schemas.MAX_ID`), so an oversized id is a 400 and never reaches asyncpg. A POST to a deck that does not exist fails on the foreign key, and `ForeignKeyError` maps to 404. 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`.
 

app/__main__.py

@@ -1,6 +1,5 @@
 import granian
 from granian.constants import Interfaces, Loops
-from granian.log import LogLevels
 
 from app.settings import settings
 
@@ -12,6 +11,6 @@
         address=settings.app_host,
         port=settings.app_port,
         interface=Interfaces.ASGI,
-        log_level=LogLevels(settings.log_level),
+        log_level=settings.log_level,
         loop=Loops.uvloop,
     ).serve()

app/api/__init__.py

@@ -0,0 +1,8 @@
+from typing import Annotated
+
+from litestar.params import PathParameter
+
+from app import schemas
+
+
+PathId = Annotated[int, PathParameter(gt=0, le=schemas.MAX_ID)]

app/api/cards.py

@@ -4,46 +4,54 @@
 import litestar
 from litestar.di import NamedDependency
 from litestar.exceptions import ValidationException
-from litestar.params import FromPath, QueryParameter
+from litestar.params import QueryParameter
 
 from app import schemas
+from app.api import PathId
 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
 
 
+TOO_MANY_CARDS: typing.Final = f"at most {schemas.MAX_CARDS_PER_REQUEST} cards per request"
+
+
 @litestar.get("/decks/{deck_id:int}/cards/")
 async def list_cards(
-    deck_id: FromPath[int],
+    deck_id: PathId,
     list_cards_use_case: NamedDependency[ListCardsUseCase],
-    after: Annotated[int | None, QueryParameter(gt=0)] = None,
+    after: Annotated[int | None, QueryParameter(gt=0, le=schemas.MAX_ID)] = None,
     limit: Annotated[int, QueryParameter(ge=1, le=schemas.MAX_PAGE_LIMIT)] = schemas.DEFAULT_PAGE_LIMIT,
 ) -> schemas.CardsPage:
     cards: typing.Final = await list_cards_use_case(deck_id=deck_id, after=after, limit=limit)
     return schemas.CardsPage.from_models(cards, limit=limit)
 
 
 @litestar.get("/cards/{card_id:int}/")
-async def get_card(card_id: FromPath[int], get_card_use_case: NamedDependency[GetCardUseCase]) -> schemas.Card:
+async def get_card(card_id: PathId, get_card_use_case: NamedDependency[GetCardUseCase]) -> schemas.Card:
     return schemas.Card.model_validate(await get_card_use_case(card_id=card_id))
 
 
 @litestar.post("/decks/{deck_id:int}/cards/")
 async def create_cards(
-    deck_id: FromPath[int],
+    deck_id: PathId,
     data: list[schemas.CardCreate],
     create_cards_use_case: NamedDependency[CreateCardsUseCase],
 ) -> schemas.Cards:
+    if len(data) > schemas.MAX_CARDS_PER_REQUEST:
+        raise ValidationException(detail=TOO_MANY_CARDS)
     return schemas.Cards.from_models(await create_cards_use_case(deck_id=deck_id, data=data))
 
 
 @litestar.put("/decks/{deck_id:int}/cards/")
 async def update_cards(
-    deck_id: FromPath[int],
+    deck_id: PathId,
     data: list[schemas.CardUpdate],
     update_cards_use_case: NamedDependency[UpdateCardsUseCase],
 ) -> schemas.Cards:
+    if len(data) > schemas.MAX_CARDS_PER_REQUEST:
+        raise ValidationException(detail=TOO_MANY_CARDS)
     if len({card.id for card in data}) != len(data):
         raise ValidationException(detail="card ids must be unique")
     return schemas.Cards.from_models(await update_cards_use_case(deck_id=deck_id, data=data))

app/api/decks.py

@@ -3,9 +3,10 @@
 
 import litestar
 from litestar.di import NamedDependency
-from litestar.params import FromPath, QueryParameter
+from litestar.params import QueryParameter
 
 from app import schemas
+from app.api import PathId
 from app.use_cases.create_deck import CreateDeckUseCase
 from app.use_cases.get_deck import GetDeckUseCase
 from app.use_cases.list_decks import ListDecksUseCase
@@ -15,21 +16,21 @@
 @litestar.get("/decks/")
 async def list_decks(
     list_decks_use_case: NamedDependency[ListDecksUseCase],
-    after: Annotated[int | None, QueryParameter(gt=0)] = None,
+    after: Annotated[int | None, QueryParameter(gt=0, le=schemas.MAX_ID)] = None,
     limit: Annotated[int, QueryParameter(ge=1, le=schemas.MAX_PAGE_LIMIT)] = schemas.DEFAULT_PAGE_LIMIT,
 ) -> schemas.DecksPage:
     decks: typing.Final = await list_decks_use_case(after=after, limit=limit)
     return schemas.DecksPage.from_models(decks, limit=limit)
 
 
 @litestar.get("/decks/{deck_id:int}/")
-async def get_deck(deck_id: FromPath[int], get_deck_use_case: NamedDependency[GetDeckUseCase]) -> schemas.DeckWithCards:
+async def get_deck(deck_id: PathId, get_deck_use_case: NamedDependency[GetDeckUseCase]) -> schemas.DeckWithCards:
     return schemas.DeckWithCards.model_validate(await get_deck_use_case(deck_id=deck_id))
 
 
 @litestar.put("/decks/{deck_id:int}/")
 async def update_deck(
-    deck_id: FromPath[int],
+    deck_id: PathId,
     data: schemas.DeckUpdate,
     update_deck_use_case: NamedDependency[UpdateDeckUseCase],
 ) -> schemas.Deck:

app/application.py

@@ -3,7 +3,7 @@
 
 import modern_di
 import modern_di_litestar
-from advanced_alchemy.exceptions import DuplicateKeyError, NotFoundError
+from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
 from lite_bootstrap import LitestarBootstrapper
 from litestar.config.app import AppConfig
 from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
@@ -25,6 +25,7 @@ def build_app() -> litestar.Litestar:
             exception_handlers={
                 DuplicateKeyError: exceptions.duplicate_key_error_handler,
                 NotFoundError: exceptions.not_found_error_handler,
+                ForeignKeyError: exceptions.not_found_error_handler,
             },
             route_handlers=[decks.ROUTER, cards.ROUTER],
             plugins=[modern_di_litestar.ModernDIPlugin(di_container, autowired_groups=[ioc.UseCases])],

app/exceptions.py

@@ -5,7 +5,7 @@
 
 
 if typing.TYPE_CHECKING:
-    from advanced_alchemy.exceptions import DuplicateKeyError, NotFoundError
+    from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
 
 
 def duplicate_key_error_handler(_: object, exc: DuplicateKeyError) -> litestar.Response[dict[str, typing.Any]]:
@@ -16,7 +16,7 @@ def duplicate_key_error_handler(_: object, exc: DuplicateKeyError) -> litestar.R
     )
 
 
-def not_found_error_handler(_: object, __: NotFoundError) -> litestar.Response[dict[str, typing.Any]]:
+def not_found_error_handler(_: object, __: NotFoundError | ForeignKeyError) -> litestar.Response[dict[str, typing.Any]]:
     return litestar.Response(
         media_type=litestar.MediaType.JSON,
         content={"detail": "Not found"},

app/models.py

@@ -17,12 +17,15 @@ class Deck(BigIntAuditBase):
     idempotency_key: orm.Mapped[uuid.UUID] = orm.mapped_column(sa.Uuid, nullable=False)
     name: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=False)
     description: orm.Mapped[str | None] = orm.mapped_column(sa.String, nullable=True)
-    cards: orm.Mapped[list[Card]] = orm.relationship("Card", lazy="noload", uselist=True)
+    cards: orm.Mapped[list[Card]] = orm.relationship("Card", lazy="noload", uselist=True, order_by="Card.id")
 
 
 class Card(BigIntAuditBase):
     __tablename__ = "cards"
-    __table_args__ = (sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc"),)
+    __table_args__ = (
+        sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc"),
+        sa.Index("ix_cards_deck_id_id", "deck_id", "id"),
+    )
 
     front: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=False)
     back: orm.Mapped[str | None] = orm.mapped_column(sa.String, nullable=True)

app/repositories.py

@@ -75,17 +75,25 @@ async def add_cards(self, deck_id: int, cards: list[schemas.CardCreate]) -> Sequ
     async def update_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> Sequence[models.Card]:
         if not cards:
             return []
-        updates: typing.Final = sa.values(
-            sa.column("id", sa.BigInteger),
-            sa.column("front", sa.String),
-            sa.column("back", sa.String),
-            sa.column("hint", sa.String),
-            name="updates",
-        ).data([(card.id, card.front, card.back, card.hint) for card in cards])
+        updates: typing.Final = (
+            sa.func.unnest(
+                sa.bindparam("ids", [card.id for card in cards], type_=postgresql.ARRAY(sa.BigInteger)),
+                sa.bindparam("fronts", [card.front for card in cards], type_=postgresql.ARRAY(sa.String)),
+                sa.bindparam("backs", [card.back for card in cards], type_=postgresql.ARRAY(sa.String)),
+                sa.bindparam("hints", [card.hint for card in cards], type_=postgresql.ARRAY(sa.String)),
+            )
+            .table_valued(
+                sa.column("id", sa.BigInteger),
+                sa.column("front", sa.String),
+                sa.column("back", sa.String),
+                sa.column("hint", sa.String),
+            )
+            .render_derived(name="updates")
+        )
         statement: typing.Final = (
             sa.update(models.Card)
             .where(models.Card.id == updates.c.id, models.Card.deck_id == deck_id)
-            .values(front=updates.c.front, back=updates.c.back, hint=updates.c.hint, updated_at=sa.func.now())
+            .values(front=updates.c.front, back=updates.c.back, hint=updates.c.hint)
             .returning(models.Card)
         )
         with wrap_sqlalchemy_exception(error_messages=DEFAULT_ERROR_MESSAGE_TEMPLATES, dialect_name="postgresql"):

app/schemas.py

@@ -1,8 +1,8 @@
 import uuid
-from typing import TYPE_CHECKING, Final, Protocol, Self
+from typing import TYPE_CHECKING, Annotated, Final, Protocol, Self
 
 import pydantic
-from pydantic import BaseModel, PositiveInt
+from pydantic import BaseModel, Field, PositiveInt
 
 
 if TYPE_CHECKING:
@@ -11,6 +11,10 @@
 
 DEFAULT_PAGE_LIMIT: Final = 50
 MAX_PAGE_LIMIT: Final = 100
+MAX_CARDS_PER_REQUEST: Final = 1000
+MAX_ID: Final = 2**63 - 1
+
+Id = Annotated[int, Field(gt=0, le=MAX_ID)]
 
 
 class HasId(Protocol):
@@ -52,7 +56,7 @@ class CardCreate(CardBase):
 
 
 class CardUpdate(CardBase):
-    id: PositiveInt
+    id: Id
 
 
 class Card(CardUpdate):
@@ -87,7 +91,7 @@ class Deck(DeckBase):
 
 
 class DeckWithCards(Deck):
-    """Deck detail view; cards are eager-loaded (selectinload) and always present."""
+    """Deck detail view; cards are eager-loaded (joinedload) and always present."""
 
     cards: list[Card]
 

app/settings.py

@@ -1,4 +1,8 @@
+import typing
+
+import pydantic
 import pydantic_settings
+from granian.log import LogLevels, log_levels_map
 from lite_bootstrap import LitestarConfig
 from sqlalchemy.engine.url import URL, make_url
 
@@ -8,7 +12,7 @@ class Settings(pydantic_settings.BaseSettings):
     service_version: str = "1.0.0"
     service_environment: str = "local"
     service_debug: bool = False
-    log_level: str = "info"
+    log_level: typing.Annotated[LogLevels, pydantic.BeforeValidator(str.lower)] = LogLevels.info
 
     db_dsn: str = "postgresql+asyncpg://postgres:password@db/postgres"
     db_replica_dsn: str = ""
@@ -49,6 +53,7 @@ def api_bootstrapper_config(self) -> LitestarConfig:
             cors_allowed_methods=self.cors_allowed_methods,
             cors_allowed_headers=self.cors_allowed_headers,
             cors_exposed_headers=self.cors_exposed_headers,
+            logging_log_level=log_levels_map[self.log_level],
             logging_buffer_capacity=self.logging_buffer_capacity,
             swagger_offline_docs=self.swagger_offline_docs,
         )

migrations/versions/2026-10-17_card_deck_id_id_index.py

@@ -0,0 +1,24 @@
+"""card deck_id id index.
+
+Revision ID: 4004ffe4656f
+Revises: 8e4a2c7b1f6d
+Create Date: 2026-10-17 12:13:24.538852
+
+"""
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "4004ffe4656f"
+down_revision = "8e4a2c7b1f6d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index("ix_cards_deck_id_id", "cards", ["deck_id", "id"], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index("ix_cards_deck_id_id", table_name="cards")