feat: make deck creation idempotent with a client-generated key

AGENTS.md

@@ -73,7 +73,10 @@ class ListDecksUseCase:
 
 - 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 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.
+- A retry runs the whole use case again. If the connection drops during `COMMIT`, the write may already have landed. A write that creates rows needs something that makes the second run harmless:
+  - `CreateDeckUseCase` takes a client-generated `idempotency_key` (`schemas.DeckCreate`, unique on `decks`). `DecksRepository.get_or_create` sends one `INSERT ... ON CONFLICT (idempotency_key) DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key RETURNING *`. On a conflict the no-op update leaves the row as it was and `RETURNING` hands it back (`DO NOTHING` would return no row), so a retry or a client resend gets the first deck. A concurrent request with the same key waits on the row lock and then gets the same deck. advanced-alchemy's own `upsert` and `get_or_upsert` do a SELECT followed by an INSERT, so they cost two round trips and lose that race.
+  - `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.
 
 `app/use_cases/` is exempt from ruff's `TC` rules because modern-di reads the dataclass field annotations at runtime.
 

app/api/decks.py

@@ -31,7 +31,7 @@ async def get_deck(
 @ROUTER.put("/decks/{deck_id}/")
 async def update_deck(
     deck_id: int,
-    data: schemas.DeckCreate,
+    data: schemas.DeckUpdate,
     update_deck_use_case: UpdateDeckUseCase = FromDI(UpdateDeckUseCase),
 ) -> schemas.Deck:
     return schemas.Deck.model_validate(await update_deck_use_case(deck_id=deck_id, data=data))

app/models.py

@@ -1,4 +1,5 @@
 import typing
+import uuid
 
 import sqlalchemy as sa
 from advanced_alchemy.base import BigIntAuditBase, orm_registry
@@ -11,7 +12,9 @@
 
 class Deck(BigIntAuditBase):
     __tablename__ = "decks"
+    __table_args__ = (sa.UniqueConstraint("idempotency_key", name="deck_idempotency_key_uc"),)
 
+    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)

app/repositories.py

@@ -4,6 +4,7 @@
 from advanced_alchemy.repository import SQLAlchemyAsyncRepository
 from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
 from sqlalchemy import orm
+from sqlalchemy.dialects import postgresql
 
 from app import models, schemas
 
@@ -21,6 +22,14 @@ class BaseRepository(SQLAlchemyAsyncRepository[models.Deck]):
     async def list_decks(self) -> Sequence[models.Deck]:
         return await self.get_many(order_by=models.Deck.id.asc())
 
+    async def get_or_create(self, data: schemas.DeckCreate) -> models.Deck:
+        statement = postgresql.insert(models.Deck).values(**data.model_dump())
+        statement = statement.on_conflict_do_update(
+            index_elements=[models.Deck.idempotency_key],
+            set_={"idempotency_key": statement.excluded.idempotency_key},
+        )
+        return (await self.repository.session.scalars(statement.returning(models.Deck))).one()
+
     async def fetch_with_cards(self, deck_id: int) -> models.Deck:
         return await self.get_one(
             models.Deck.id == deck_id,

app/schemas.py

@@ -1,3 +1,4 @@
+import uuid
 from typing import TYPE_CHECKING, Self
 
 import pydantic
@@ -45,6 +46,10 @@ class DeckBase(Base):
 
 
 class DeckCreate(DeckBase):
+    idempotency_key: uuid.UUID
+
+
+class DeckUpdate(DeckBase):
     pass
 
 

app/use_cases/create_deck.py

@@ -15,6 +15,6 @@ class CreateDeckUseCase:
     @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())
+            deck: typing.Final = await self.decks_repository.get_or_create(data)
             await self.transaction.commit()
         return deck

app/use_cases/update_deck.py

@@ -13,7 +13,7 @@ class UpdateDeckUseCase:
     decks_repository: DecksRepository
 
     @postgres_retry
-    async def __call__(self, *, deck_id: int, data: schemas.DeckCreate) -> models.Deck:
+    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)
             await self.transaction.commit()

migrations/versions/2026-10-16_deck_idempotency_key.py

@@ -0,0 +1,31 @@
+"""deck idempotency key.
+
+Revision ID: 5d1c0f3a9b2e
+Revises: abb62721e019
+Create Date: 2026-10-16 20:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "5d1c0f3a9b2e"
+down_revision = "abb62721e019"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "decks",
+        sa.Column("idempotency_key", sa.Uuid(), nullable=False, server_default=sa.text("gen_random_uuid()")),
+    )
+    op.alter_column("decks", "idempotency_key", server_default=None)
+    op.create_unique_constraint("deck_idempotency_key_uc", "decks", ["idempotency_key"])
+
+
+def downgrade() -> None:
+    op.drop_constraint("deck_idempotency_key_uc", "decks", type_="unique")
+    op.drop_column("decks", "idempotency_key")

pyproject.toml

@@ -64,6 +64,7 @@ ignore = [
     "CPY001", # Missing copyright notice at top of file
 ]
 isort.lines-after-imports = 2
+flake8-type-checking.runtime-evaluated-base-classes = ["pydantic.BaseModel", "advanced_alchemy.base.BigIntAuditBase"]
 isort.no-lines-before = ["standard-library", "local-folder"]
 
 [tool.ruff.lint.extend-per-file-ignores]

tests/test_create_deck.py

@@ -0,0 +1,51 @@
+import asyncio
+import typing
+import uuid
+
+import sqlalchemy as sa
+from db_retry import Transaction
+
+from app import models, schemas
+from app.repositories import DecksRepository
+from app.resources.db import create_primary_sa_engine, create_session
+from app.use_cases.create_deck import CreateDeckUseCase
+
+
+if typing.TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncEngine
+
+
+async def _wait_until_blocked_on_a_lock(engine: AsyncEngine, pid: int) -> None:
+    wait_event_type = None
+    while wait_event_type != "Lock":
+        await asyncio.sleep(0.01)
+        async with engine.connect() as connection:
+            wait_event_type = await connection.scalar(
+                sa.text("SELECT wait_event_type FROM pg_stat_activity WHERE pid = :pid"), {"pid": pid}
+            )
+
+
+async def test_create_deck_returns_the_deck_a_concurrent_request_created_with_the_same_key() -> None:
+    engine: typing.Final = create_primary_sa_engine()
+    idempotency_key: typing.Final = uuid.uuid4()
+    try:
+        async with engine.connect() as concurrent_request:
+            await concurrent_request.execute(
+                sa.insert(models.Deck).values(idempotency_key=idempotency_key, name="first")
+            )
+            session: typing.Final = create_session(engine)
+            pid: typing.Final = await (await session.connection()).scalar(sa.text("SELECT pg_backend_pid()"))
+            use_case: typing.Final = CreateDeckUseCase(
+                transaction=Transaction(session=session), decks_repository=DecksRepository(session=session)
+            )
+            pending: typing.Final = asyncio.create_task(
+                use_case(data=schemas.DeckCreate(idempotency_key=idempotency_key, name="second"))
+            )
+            await asyncio.wait_for(_wait_until_blocked_on_a_lock(engine, pid), timeout=5)
+            await concurrent_request.commit()
+            deck: typing.Final = await pending
+        assert deck.name == "first"
+    finally:
+        async with engine.begin() as connection:
+            await connection.execute(sa.delete(models.Deck).where(models.Deck.idempotency_key == idempotency_key))
+        await engine.dispose()

tests/test_decks.py

@@ -1,3 +1,4 @@
+import uuid
 from typing import TYPE_CHECKING
 
 import pytest
@@ -89,6 +90,7 @@ async def test_post_decks(
     response = await client.post(
         "/api/decks/",
         json={
+            "idempotency_key": str(uuid.uuid4()),
             "name": name,
             "description": description,
         },
@@ -149,3 +151,21 @@ async def test_put_decks(client: AsyncClient, name: str, description: str) -> No
     result = response.json()
     assert name == result["name"]
     assert description == result["description"]
+
+
+async def test_post_decks_requires_an_idempotency_key(client: AsyncClient) -> None:
+    response = await client.post("/api/decks/", json={"name": "test deck"})
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
+
+
+async def test_post_decks_twice_with_one_idempotency_key_creates_one_deck(client: AsyncClient) -> None:
+    idempotency_key = str(uuid.uuid4())
+
+    first = await client.post("/api/decks/", json={"idempotency_key": idempotency_key, "name": "first"})
+    second = await client.post("/api/decks/", json={"idempotency_key": idempotency_key, "name": "second"})
+
+    assert first.status_code == status.HTTP_200_OK
+    assert second.status_code == status.HTTP_200_OK
+    assert second.json() == first.json()
+    response = await client.get("/api/decks/")
+    assert [item["name"] for item in response.json()["items"]] == ["first"]