Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
- `dynamic_engine` — `Scope.REQUEST`; `choose_sa_engine` returns the read engine for `GET`/`HEAD` requests, the primary otherwise (including when there is no request). GET handlers cannot write, and a read right after a write may see replica lag.
- `session` — `Scope.REQUEST`, bound to `dynamic_engine`, finalized by `close_session`.
- `transaction` — `Scope.REQUEST`, a `db_retry.Transaction` over `session`. Leaving its block rolls back anything uncommitted and closes the session.
- `autocommit_transaction` — the same `Transaction`, opened at `AUTOCOMMIT` (`autocommit_isolation_level`). `CreateDeckUseCase` and `UpdateDeckUseCase` use it because each sends exactly one statement, which Postgres runs atomically on its own. The asyncpg adapter then sends no BEGIN or COMMIT, so those writes cost one round trip instead of three. A use case with two or more statements, or one that relies on rolling back after a check, as `UpdateCardsUseCase` does, keeps `transaction`. The `db_session` fixture overrides `autocommit_isolation_level` to `None`, because switching the shared test connection to `AUTOCOMMIT` would roll back its outer transaction. `tests/test_autocommit_writes.py` runs both use cases against a real engine instead.
- `autocommit_transaction` — the same `Transaction`, opened at `AUTOCOMMIT` (`autocommit_isolation_level`). `CreateDeckUseCase` and `UpdateDeckUseCase` use it because each sends exactly one statement, which Postgres runs atomically on its own. The asyncpg adapter then sends no BEGIN or COMMIT, so those writes cost one round trip instead of three. A use case with two or more statements, or one that relies on rolling back after a check, as `UpdateCardsUseCase` does, keeps `transaction`. The `db_session` fixture overrides `autocommit_isolation_level` to `None`, because switching the shared test connection to `AUTOCOMMIT` would roll back its outer transaction. `tests/test_autocommit_writes.py` runs both use cases against a real engine instead, through the `committing_engine` fixture. `tests/test_card_locking.py` uses the same fixture, because a lock wait needs separate connections; both delete the rows they commit.

`Repositories` (`Scope.REQUEST`): `decks_repository` and `cards_repository`, on `Database.session` with `auto_commit=False`. The use case commits through `transaction.commit()`.

Expand Down Expand Up @@ -89,7 +89,7 @@ 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 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/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 `WITH locked AS (SELECT id FROM cards WHERE id = ANY(:ids) AND deck_id = :deck_id ORDER BY id FOR UPDATE) UPDATE cards ... FROM unnest(:ids, :fronts, :backs, :hints) AS updates(...), locked WHERE cards.id = updates.id AND cards.id = locked.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. Every updated row must match a `locked` row, and the CTE locks its rows in id order (lazily, one before each update), so concurrent PUTs on overlapping cards always lock them in the same order and cannot deadlock. Without it the lock order follows the plan, which can be the order the client sent; `tests/test_card_locking.py` forces that interleaving. `card_deck_id_front_uc` is `DEFERRABLE INITIALLY IMMEDIATE`, so uniqueness is checked at the end of the statement and one PUT can swap two cards' fronts. 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`.

Expand Down
2 changes: 1 addition & 1 deletion app/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ class Deck(BigIntAuditBase):
class Card(BigIntAuditBase):
__tablename__ = "cards"
__table_args__ = (
sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc"),
sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc", deferrable=True, initially="IMMEDIATE"),
sa.Index("ix_cards_deck_id_id", "deck_id", "id"),
)

Expand Down
12 changes: 10 additions & 2 deletions app/repositories.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,9 +75,17 @@ 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 []
ids: typing.Final = sa.bindparam("ids", [card.id for card in cards], type_=postgresql.ARRAY(sa.BigInteger))
locked: typing.Final = (
sa.select(models.Card.id)
.where(models.Card.id == sa.any_(ids), models.Card.deck_id == deck_id)
.order_by(models.Card.id)
.with_for_update()
.cte("locked")
)
updates: typing.Final = (
sa.func.unnest(
sa.bindparam("ids", [card.id for card in cards], type_=postgresql.ARRAY(sa.BigInteger)),
ids,
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)),
Expand All @@ -92,7 +100,7 @@ async def update_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> S
)
statement: typing.Final = (
sa.update(models.Card)
.where(models.Card.id == updates.c.id, models.Card.deck_id == deck_id)
.where(models.Card.id == updates.c.id, models.Card.id == locked.c.id, models.Card.deck_id == deck_id)
.values(front=updates.c.front, back=updates.c.back, hint=updates.c.hint)
.returning(models.Card)
)
Expand Down
28 changes: 28 additions & 0 deletions migrations/versions/2026-10-17_deferrable_card_front_unique.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
"""deferrable card front unique.

Revision ID: 456d08f790fb
Revises: 4004ffe4656f
Create Date: 2026-10-17 15:00:00.000000

"""

from alembic import op


# revision identifiers, used by Alembic.
revision = "456d08f790fb"
down_revision = "4004ffe4656f"
branch_labels = None
depends_on = None


def upgrade() -> None:
op.drop_constraint("card_deck_id_front_uc", "cards", type_="unique")
op.create_unique_constraint(
"card_deck_id_front_uc", "cards", ["deck_id", "front"], deferrable=True, initially="IMMEDIATE"
)


def downgrade() -> None:
op.drop_constraint("card_deck_id_front_uc", "cards", type_="unique")
op.create_unique_constraint("card_deck_id_front_uc", "cards", ["deck_id", "front"])
13 changes: 12 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
import fastapi
import modern_di
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession


SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
Expand Down Expand Up @@ -92,3 +92,14 @@ def record(*args: object) -> None:
yield recorded
finally:
event.remove(sync_connection, "before_cursor_execute", record)


@pytest.fixture
async def committing_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
engine = create_primary_sa_engine()
di_container.override(ioc.Database.dynamic_engine, engine)
try:
yield engine
finally:
di_container.reset_override(ioc.Database.dynamic_engine)
await engine.dispose()
22 changes: 5 additions & 17 deletions tests/test_autocommit_writes.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,24 +9,12 @@
from sqlalchemy import event

from app import ioc, models, schemas
from app.resources.db import create_primary_sa_engine


if typing.TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncEngine


@pytest.fixture
async def primary_engine(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncEngine]:
engine = create_primary_sa_engine()
di_container.override(ioc.Database.dynamic_engine, engine)
try:
yield engine
finally:
di_container.reset_override(ioc.Database.dynamic_engine)
await engine.dispose()


@contextlib.contextmanager
def record_in_transaction(engine: AsyncEngine) -> typing.Iterator[list[bool]]:
in_transaction: typing.Final[list[bool]] = []
Expand All @@ -44,25 +32,25 @@ def record(connection: sa.Connection, *_: object) -> None:


async def test_create_deck_sends_no_begin_or_commit(
di_container: modern_di.Container, primary_engine: AsyncEngine
di_container: modern_di.Container, committing_engine: AsyncEngine
) -> None:
key: typing.Final = uuid.uuid4()
try:
with record_in_transaction(primary_engine) as in_transaction:
with record_in_transaction(committing_engine) as in_transaction:
async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
use_case = container.resolve_provider(ioc.UseCases.create_deck_use_case)
await use_case(data=schemas.DeckCreate(name="autocommit", idempotency_key=key))
finally:
async with primary_engine.begin() as connection:
async with committing_engine.begin() as connection:
await connection.execute(sa.delete(models.Deck).where(models.Deck.idempotency_key == key))

assert in_transaction == [False]


async def test_update_deck_sends_no_begin_or_commit(
di_container: modern_di.Container, primary_engine: AsyncEngine
di_container: modern_di.Container, committing_engine: AsyncEngine
) -> None:
with record_in_transaction(primary_engine) as in_transaction:
with record_in_transaction(committing_engine) as in_transaction:
async with di_container.build_child_container(scope=modern_di.Scope.REQUEST) as container:
use_case = container.resolve_provider(ioc.UseCases.update_deck_use_case)
with pytest.raises(NotFoundError):
Expand Down
73 changes: 73 additions & 0 deletions tests/test_card_locking.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
import asyncio
import typing
import uuid

import sqlalchemy as sa
from fastapi import status

from app import models


if typing.TYPE_CHECKING:
from httpx import AsyncClient, Response
from sqlalchemy.ext.asyncio import AsyncEngine


async def wait_for_lock_waiters(engine: AsyncEngine, count: int, request: asyncio.Task[typing.Any]) -> None:
statement: typing.Final = sa.text(
"SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() AND wait_event_type = 'Lock'"
)
async with asyncio.timeout(5):
while not request.done():
async with engine.connect() as connection:
if await connection.scalar(statement) >= count:
return
await asyncio.sleep(0.01)


async def test_concurrent_card_updates_in_opposite_orders_do_not_deadlock(
client: AsyncClient, committing_engine: AsyncEngine
) -> None:
async with committing_engine.begin() as connection:
deck_id = await connection.scalar(
sa.insert(models.Deck).values(name="locking", idempotency_key=uuid.uuid4()).returning(models.Deck.id)
)
card1, card2, card3 = (
await connection.scalars(
sa.insert(models.Card)
.values([{"deck_id": deck_id, "front": f"card{i}"} for i in range(1, 4)])
.returning(models.Card.id)
)
).all()
url: typing.Final = f"/api/decks/{deck_id}/cards/"
requests: typing.Final[list[asyncio.Task[Response]]] = []
try:
async with committing_engine.connect() as blocker:
await blocker.execute(sa.text("SELECT 1 FROM cards WHERE id = :id FOR UPDATE"), {"id": card3})
requests.append(
asyncio.create_task(
client.put(
url,
json=[
{"id": card1, "front": "a1"},
{"id": card3, "front": "a3"},
{"id": card2, "front": "a2"},
],
)
)
)
await wait_for_lock_waiters(committing_engine, 1, requests[0])
requests.append(
asyncio.create_task(client.put(url, json=[{"id": card2, "front": "b2"}, {"id": card1, "front": "b1"}]))
)
await wait_for_lock_waiters(committing_engine, 2, requests[1])
responses = await asyncio.gather(*requests)
finally:
for request in requests:
request.cancel()
await asyncio.gather(*requests, return_exceptions=True)
async with committing_engine.begin() as connection:
await connection.execute(sa.delete(models.Card).where(models.Card.deck_id == deck_id))
await connection.execute(sa.delete(models.Deck).where(models.Deck.id == deck_id))

assert [response.status_code for response in responses] == [status.HTTP_200_OK, status.HTTP_200_OK]
12 changes: 12 additions & 0 deletions tests/test_cards.py
Original file line number Diff line number Diff line change
Expand Up @@ -321,3 +321,15 @@ async def test_card_endpoints_reject_ids_out_of_bigint_range(
) -> None:
response = await client.request(method, path, json=body, params=params)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT


async def test_update_cards_swaps_fronts(client: AsyncClient) -> None:
deck = await factories.DeckModelFactory.create_async()
card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)

response = await client.put(
f"/api/decks/{deck.id}/cards/",
json=[{"id": card1.id, "front": card2.front}, {"id": card2.id, "front": card1.front}],
)
assert response.status_code == status.HTTP_200_OK
assert [x["front"] for x in response.json()["items"]] == [card2.front, card1.front]
Loading