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

AGENTS.md

@@ -84,13 +84,13 @@ 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_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. `schemas.CardUpdates` rejects duplicate ids with a 422 before any query runs. A statement that can hit a unique constraint runs inside `wrap_sqlalchemy_exception` so that it still raises `DuplicateKeyError`.
+- `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` 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. `schemas.CardCreates` and `schemas.CardUpdates` cap a request at `MAX_CARDS_PER_REQUEST` cards, and `schemas.CardUpdates` rejects duplicate ids, both with a 422 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 422 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`.
 - `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`.
 
 ### Settings
 
-`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`.
+`app/settings.py` — `pydantic_settings.BaseSettings`. Env vars are unprefixed (`DB_DSN`, `DB_REPLICA_DSN`, `SERVICE_DEBUG`, `SERVICE_ENVIRONMENT`, `LOG_LEVEL`, `APP_HOST`, `APP_PORT`, `OPENTELEMETRY_ENDPOINT`, `SENTRY_DSN`, `CORS_ALLOWED_ORIGINS`, ...). `api_bootstrapper_config` produces a `FastAPIConfig` for `lite-bootstrap`. `LOG_LEVEL` is case-insensitive and sets both Granian's log level and the app's root logger.
 
 ### Tests
 

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 @@
+import typing
+
+import fastapi
+
+from app import schemas
+
+
+PathId = typing.Annotated[int, fastapi.Path(gt=0, le=schemas.MAX_ID)]

app/api/cards.py

@@ -4,6 +4,7 @@
 from modern_di_fastapi import FromDI
 
 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
@@ -15,8 +16,8 @@
 
 @ROUTER.get("/decks/{deck_id}/cards/")
 async def list_cards(
-    deck_id: int,
-    after: typing.Annotated[int | None, fastapi.Query(gt=0)] = None,
+    deck_id: PathId,
+    after: typing.Annotated[int | None, fastapi.Query(gt=0, le=schemas.MAX_ID)] = None,
     limit: typing.Annotated[int, fastapi.Query(ge=1, le=schemas.MAX_PAGE_LIMIT)] = schemas.DEFAULT_PAGE_LIMIT,
     list_cards_use_case: ListCardsUseCase = FromDI(ListCardsUseCase),
 ) -> schemas.CardsPage:
@@ -26,24 +27,24 @@ async def list_cards(
 
 @ROUTER.get("/cards/{card_id}/")
 async def get_card(
-    card_id: int,
+    card_id: PathId,
     get_card_use_case: GetCardUseCase = FromDI(GetCardUseCase),
 ) -> schemas.Card:
     return schemas.Card.model_validate(await get_card_use_case(card_id=card_id))
 
 
 @ROUTER.post("/decks/{deck_id}/cards/")
 async def create_cards(
-    deck_id: int,
-    data: list[schemas.CardCreate],
+    deck_id: PathId,
+    data: schemas.CardCreates,
     create_cards_use_case: CreateCardsUseCase = FromDI(CreateCardsUseCase),
 ) -> schemas.Cards:
     return schemas.Cards.from_models(await create_cards_use_case(deck_id=deck_id, data=data))
 
 
 @ROUTER.put("/decks/{deck_id}/cards/")
 async def update_cards(
-    deck_id: int,
+    deck_id: PathId,
     data: schemas.CardUpdates,
     update_cards_use_case: UpdateCardsUseCase = FromDI(UpdateCardsUseCase),
 ) -> schemas.Cards:

app/api/decks.py

@@ -4,6 +4,7 @@
 from modern_di_fastapi import FromDI
 
 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,7 +16,7 @@
 
 @ROUTER.get("/decks/")
 async def list_decks(
-    after: typing.Annotated[int | None, fastapi.Query(gt=0)] = None,
+    after: typing.Annotated[int | None, fastapi.Query(gt=0, le=schemas.MAX_ID)] = None,
     limit: typing.Annotated[int, fastapi.Query(ge=1, le=schemas.MAX_PAGE_LIMIT)] = schemas.DEFAULT_PAGE_LIMIT,
     list_decks_use_case: ListDecksUseCase = FromDI(ListDecksUseCase),
 ) -> schemas.DecksPage:
@@ -25,15 +26,15 @@ async def list_decks(
 
 @ROUTER.get("/decks/{deck_id}/")
 async def get_deck(
-    deck_id: int,
+    deck_id: PathId,
     get_deck_use_case: GetDeckUseCase = FromDI(GetDeckUseCase),
 ) -> schemas.DeckWithCards:
     return schemas.DeckWithCards.model_validate(await get_deck_use_case(deck_id=deck_id))
 
 
 @ROUTER.put("/decks/{deck_id}/")
 async def update_deck(
-    deck_id: int,
+    deck_id: PathId,
     data: schemas.DeckUpdate,
     update_deck_use_case: UpdateDeckUseCase = FromDI(UpdateDeckUseCase),
 ) -> schemas.Deck:

app/application.py

@@ -3,7 +3,7 @@
 
 import modern_di
 import modern_di_fastapi
-from advanced_alchemy.exceptions import DuplicateKeyError, NotFoundError
+from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
 from lite_bootstrap import FastAPIBootstrapper
 from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
 
@@ -39,4 +39,8 @@ def build_app() -> fastapi.FastAPI:
         NotFoundError,
         exceptions.not_found_error_handler,  # ty: ignore[invalid-argument-type]
     )
+    app.add_exception_handler(
+        ForeignKeyError,
+        exceptions.not_found_error_handler,  # ty: ignore[invalid-argument-type]
+    )
     return app

app/exceptions.py

@@ -5,7 +5,7 @@
 
 
 if TYPE_CHECKING:
-    from advanced_alchemy.exceptions import DuplicateKeyError, NotFoundError
+    from advanced_alchemy.exceptions import DuplicateKeyError, ForeignKeyError, NotFoundError
     from starlette.requests import Request
 
 
@@ -16,7 +16,7 @@ async def duplicate_key_error_handler(_: Request, exc: DuplicateKeyError) -> JSO
     )
 
 
-async def not_found_error_handler(_: Request, __: NotFoundError) -> JSONResponse:
+async def not_found_error_handler(_: Request, __: NotFoundError | ForeignKeyError) -> JSONResponse:
     return JSONResponse(
         status_code=status.HTTP_404_NOT_FOUND,
         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

@@ -2,7 +2,7 @@
 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
 
 
 def _ensure_unique_ids(cards: list[CardUpdate]) -> list[CardUpdate]:
@@ -62,7 +66,10 @@ def _ensure_unique_ids(cards: list[CardUpdate]) -> list[CardUpdate]:
     return cards
 
 
-CardUpdates = Annotated[list[CardUpdate], pydantic.AfterValidator(_ensure_unique_ids)]
+CardCreates = Annotated[list[CardCreate], Field(max_length=MAX_CARDS_PER_REQUEST)]
+CardUpdates = Annotated[
+    list[CardUpdate], Field(max_length=MAX_CARDS_PER_REQUEST), pydantic.AfterValidator(_ensure_unique_ids)
+]
 
 
 class Card(CardUpdate):
@@ -97,7 +104,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 FastAPIConfig
 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 = ""
@@ -47,6 +51,7 @@ def api_bootstrapper_config(self) -> FastAPIConfig:
             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")

pyproject.toml

@@ -69,6 +69,9 @@ flake8-type-checking.runtime-evaluated-base-classes = ["pydantic.BaseModel", "ad
 isort.no-lines-before = ["standard-library", "local-folder"]
 
 [tool.ruff.lint.extend-per-file-ignores]
+"app/api/*.py" = [
+    "TC", # FastAPI reads route annotations at runtime
+]
 "app/resources/*.py" = [
     "TC", # modern-di reads DI creator annotations at runtime
 ]