feat: make deck creation idempotent with a client-generated key (#90)

* test: require an idempotency key on POST /decks/ and replay it

Posting twice with one key must return the first deck and leave a single row.

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

CreateDeckUseCase looks the key up inside its transaction and returns the existing deck, so a postgres_retry rerun after a COMMIT that landed, or a client resend, does not create a second deck. PUT /decks/ takes the new DeckUpdate schema without the key.

* test: a concurrent create with the same idempotency key returns one deck

A connection holds an uncommitted insert with the key while CreateDeckUseCase runs. The lookup version fails with a duplicate-key error once that insert commits.

* perf: create a deck in one INSERT ... ON CONFLICT DO UPDATE RETURNING

The no-op update on idempotency_key returns the existing row on a replay, and a concurrent request with the same key waits on the row lock instead of failing on the unique constraint.

Author: lesnik512

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()