From 942fe8c521c089eb3dab207133bf8c295fe65c80 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 22:17:48 +0300 Subject: [PATCH 1/4] 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. --- tests/test_decks.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/test_decks.py b/tests/test_decks.py index 5686fa0..b607eb9 100644 --- a/tests/test_decks.py +++ b/tests/test_decks.py @@ -1,3 +1,4 @@ +import uuid from typing import TYPE_CHECKING import pytest @@ -78,6 +79,7 @@ async def test_post_decks( response = await client.post( "/api/decks/", json={ + "idempotency_key": str(uuid.uuid4()), "name": name, "description": description, }, @@ -138,3 +140,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"] From 0d825c596641e263a3ab8759badd580bbad66dc2 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 22:17:48 +0300 Subject: [PATCH 2/4] 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. --- AGENTS.md | 5 ++- app/api/decks.py | 2 +- app/models.py | 3 ++ app/repositories.py | 4 +++ app/schemas.py | 5 +++ app/use_cases/create_deck.py | 5 +-- app/use_cases/update_deck.py | 2 +- .../2026-10-16_deck_idempotency_key.py | 31 +++++++++++++++++++ pyproject.toml | 1 + 9 files changed, 53 insertions(+), 5 deletions(-) create mode 100644 migrations/versions/2026-10-16_deck_idempotency_key.py diff --git a/AGENTS.md b/AGENTS.md index 8c5fb9d..a525cc4 100644 --- a/AGENTS.md +++ b/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`). It looks the key up inside the transaction and returns the existing deck when it finds one, so a retry, or a client resending the request, gets the first deck back. Two concurrent requests with one key can still both miss the lookup; the second then fails on the unique constraint as a duplicate-key error. + - `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. diff --git a/app/api/decks.py b/app/api/decks.py index 29eb8f2..83d8a4e 100644 --- a/app/api/decks.py +++ b/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)) diff --git a/app/models.py b/app/models.py index d3a90e3..3551f82 100644 --- a/app/models.py +++ b/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) diff --git a/app/repositories.py b/app/repositories.py index af6b577..2801b85 100644 --- a/app/repositories.py +++ b/app/repositories.py @@ -8,6 +8,7 @@ if TYPE_CHECKING: + import uuid from collections.abc import Sequence @@ -17,6 +18,9 @@ class BaseRepository(SQLAlchemyAsyncRepository[models.Deck]): repository_type = BaseRepository + async def fetch_by_idempotency_key(self, idempotency_key: uuid.UUID) -> models.Deck | None: + return await self.get_one_or_none(idempotency_key=idempotency_key) + async def fetch_with_cards(self, deck_id: int) -> models.Deck: return await self.get_one( models.Deck.id == deck_id, diff --git a/app/schemas.py b/app/schemas.py index 2dc630d..42a7fc5 100644 --- a/app/schemas.py +++ b/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 diff --git a/app/use_cases/create_deck.py b/app/use_cases/create_deck.py index 80713d3..483b2e9 100644 --- a/app/use_cases/create_deck.py +++ b/app/use_cases/create_deck.py @@ -1,5 +1,4 @@ import dataclasses -import typing from db_retry import Transaction, postgres_retry @@ -15,6 +14,8 @@ 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 = await self.decks_repository.fetch_by_idempotency_key(data.idempotency_key) + if deck is None: + deck = await self.decks_repository.create(data.model_dump()) await self.transaction.commit() return deck diff --git a/app/use_cases/update_deck.py b/app/use_cases/update_deck.py index 303238a..fff4fa5 100644 --- a/app/use_cases/update_deck.py +++ b/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() diff --git a/migrations/versions/2026-10-16_deck_idempotency_key.py b/migrations/versions/2026-10-16_deck_idempotency_key.py new file mode 100644 index 0000000..9a6e76e --- /dev/null +++ b/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") diff --git a/pyproject.toml b/pyproject.toml index 440d757..3b9dc8c 100644 --- a/pyproject.toml +++ b/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] From 1ed7f0f9426087059066118204dcd0def6dab036 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 22:56:01 +0300 Subject: [PATCH 3/4] 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. --- tests/test_create_deck.py | 51 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) create mode 100644 tests/test_create_deck.py diff --git a/tests/test_create_deck.py b/tests/test_create_deck.py new file mode 100644 index 0000000..8026741 --- /dev/null +++ b/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() From d100eca85b0e494b180eeccbff995ef1f8e22893 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Fri, 16 Oct 2026 22:56:01 +0300 Subject: [PATCH 4/4] 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. --- AGENTS.md | 2 +- app/repositories.py | 11 ++++++++--- app/use_cases/create_deck.py | 5 ++--- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a525cc4..c769ed6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,7 +74,7 @@ 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. 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`). It looks the key up inside the transaction and returns the existing deck when it finds one, so a retry, or a client resending the request, gets the first deck back. Two concurrent requests with one key can still both miss the lookup; the second then fails on the unique constraint as a duplicate-key error. + - `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. diff --git a/app/repositories.py b/app/repositories.py index 2801b85..87f7535 100644 --- a/app/repositories.py +++ b/app/repositories.py @@ -3,12 +3,12 @@ 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 if TYPE_CHECKING: - import uuid from collections.abc import Sequence @@ -18,8 +18,13 @@ class BaseRepository(SQLAlchemyAsyncRepository[models.Deck]): repository_type = BaseRepository - async def fetch_by_idempotency_key(self, idempotency_key: uuid.UUID) -> models.Deck | None: - return await self.get_one_or_none(idempotency_key=idempotency_key) + 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( diff --git a/app/use_cases/create_deck.py b/app/use_cases/create_deck.py index 483b2e9..cc35dae 100644 --- a/app/use_cases/create_deck.py +++ b/app/use_cases/create_deck.py @@ -1,4 +1,5 @@ import dataclasses +import typing from db_retry import Transaction, postgres_retry @@ -14,8 +15,6 @@ class CreateDeckUseCase: @postgres_retry async def __call__(self, *, data: schemas.DeckCreate) -> models.Deck: async with self.transaction: - deck = await self.decks_repository.fetch_by_idempotency_key(data.idempotency_key) - if deck is None: - deck = 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