diff --git a/AGENTS.md b/AGENTS.md index 43d5efa..9e5f2be 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -43,7 +43,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f `Database`: - `database_engine` — singleton-ish primary `AsyncEngine` (`create_primary_sa_engine`, `close_sa_engine` finalizer). -- `database_read_engine` — singleton-ish read `AsyncEngine` built from `DB_REPLICA_DSN`, or from `DB_DSN` when it is unset. It runs in `AUTOCOMMIT` (no BEGIN/ROLLBACK per request) and sets `default_transaction_read_only=on`, so Postgres rejects writes on it. Each statement sees its own snapshot, so a multi-query read (e.g. `fetch_with_cards`) can see a write that commits in between. Without a replica it is a second pool on the primary, so a process can hold up to `2 × db_pool_size` connections. +- `database_read_engine` — singleton-ish read `AsyncEngine` built from `DB_REPLICA_DSN`, or from `DB_DSN` when it is unset. It runs in `AUTOCOMMIT` (no BEGIN/ROLLBACK per request) and sets `default_transaction_read_only=on`, so Postgres rejects writes on it. Each statement sees its own snapshot, so a read that runs several queries can see a write that commits in between. Load related rows in the same query (`fetch_with_cards` uses `joinedload`). Without a replica it is a second pool on the primary, so a process can hold up to `2 × db_pool_size` connections. - `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. @@ -84,7 +84,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`, `update`, `create_many`, `upsert_many`). +- `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` and `upsert_cards` send `INSERT ... ON CONFLICT ... RETURNING`. 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`. @@ -106,7 +106,7 @@ class ListDecksUseCase: - Ruff is configured with `select = ["ALL"]` and a curated ignore list in `pyproject.toml`. Don't sprinkle `# noqa`; prefer fixing or extending the project ignore list if a rule is genuinely wrong for the codebase. - Routes convert ORM objects to schemas explicitly, never via `typing.cast`. Single objects: `schemas.X.model_validate(instance)`. Collections: `schemas.Xs.from_models(objects)` (the `Collection[T]` seam in `app/schemas.py`). Both rely on `from_attributes=True`. -- Deck responses are deliberately two-shaped: `list_decks`/`create_deck`/`update_deck` return the light `schemas.Deck` (no `cards`), while `get_deck` returns `schemas.DeckWithCards`. The split mirrors loading — lists/writes use `noload` (no cards query), detail uses `selectinload` via `fetch_with_cards` — so the type states exactly what each endpoint loads. +- Deck responses are deliberately two-shaped: `list_decks`/`create_deck`/`update_deck` return the light `schemas.Deck` (no `cards`), while `get_deck` returns `schemas.DeckWithCards`. The split mirrors loading — lists/writes use `noload` (no cards query), detail uses `joinedload` via `fetch_with_cards` — so the type states exactly what each endpoint loads. - Line length is 120. ## Agent skills diff --git a/app/repositories.py b/app/repositories.py index 9accdc1..7f0cf21 100644 --- a/app/repositories.py +++ b/app/repositories.py @@ -1,7 +1,10 @@ import operator +import typing from typing import TYPE_CHECKING -from advanced_alchemy.repository import SQLAlchemyAsyncRepository +import sqlalchemy as sa +from advanced_alchemy.exceptions import NotFoundError, wrap_sqlalchemy_exception +from advanced_alchemy.repository import DEFAULT_ERROR_MESSAGE_TEMPLATES, SQLAlchemyAsyncRepository from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService from sqlalchemy import orm from sqlalchemy.dialects import postgresql @@ -33,9 +36,19 @@ async def get_or_create(self, data: schemas.DeckCreate) -> models.Deck: async def fetch_with_cards(self, deck_id: int) -> models.Deck: return await self.get_one( models.Deck.id == deck_id, - load=[orm.selectinload(models.Deck.cards)], + load=[orm.joinedload(models.Deck.cards)], + uniquify=True, ) + async def update_deck(self, deck_id: int, data: schemas.DeckUpdate) -> models.Deck: + statement: typing.Final = ( + sa.update(models.Deck).where(models.Deck.id == deck_id).values(**data.model_dump()).returning(models.Deck) + ) + deck: typing.Final = (await self.repository.session.scalars(statement)).one_or_none() + if deck is None: + raise NotFoundError + return deck + class CardsRepository(SQLAlchemyAsyncRepositoryService[models.Card]): class BaseRepository(SQLAlchemyAsyncRepository[models.Card]): @@ -50,7 +63,19 @@ async def add_cards(self, deck_id: int, cards: list[schemas.CardCreate]) -> Sequ return await self.create_many([models.Card(**card.model_dump(), deck_id=deck_id) for card in cards]) async def upsert_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> Sequence[models.Card]: - upserted = await self.upsert_many( - [models.Card(**card.model_dump(), deck_id=deck_id) for card in cards], + if not cards: + return [] + statement = postgresql.insert(models.Card).values([{**card.model_dump(), "deck_id": deck_id} for card in cards]) + statement = statement.on_conflict_do_update( + index_elements=[models.Card.id], + set_={ + "front": statement.excluded.front, + "back": statement.excluded.back, + "hint": statement.excluded.hint, + "deck_id": statement.excluded.deck_id, + "updated_at": sa.func.now(), + }, ) + with wrap_sqlalchemy_exception(error_messages=DEFAULT_ERROR_MESSAGE_TEMPLATES, dialect_name="postgresql"): + upserted: typing.Final = (await self.repository.session.scalars(statement.returning(models.Card))).all() return sorted(upserted, key=operator.attrgetter("id")) diff --git a/app/use_cases/update_deck.py b/app/use_cases/update_deck.py index fff4fa5..c3a60e6 100644 --- a/app/use_cases/update_deck.py +++ b/app/use_cases/update_deck.py @@ -15,6 +15,6 @@ class UpdateDeckUseCase: @postgres_retry 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) + deck: typing.Final = await self.decks_repository.update_deck(deck_id, data) await self.transaction.commit() return deck diff --git a/tests/conftest.py b/tests/conftest.py index 41f1c15..5c38fae 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,6 +5,7 @@ from asgi_lifespan import LifespanManager from httpx import ASGITransport, AsyncClient from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory +from sqlalchemy import event from app import ioc from app.application import build_app @@ -14,9 +15,13 @@ if typing.TYPE_CHECKING: import fastapi import modern_di + from sqlalchemy.engine import Connection from sqlalchemy.ext.asyncio import AsyncSession +SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT") + + @pytest.fixture async def app() -> typing.AsyncIterator[fastapi.FastAPI]: app_ = build_app() @@ -68,3 +73,20 @@ async def set_async_session_in_base_sqlalchemy_factory( yield finally: SQLAlchemyFactory.__async_session__ = None + + +@pytest.fixture +def statements(db_session: AsyncSession) -> typing.Iterator[list[str]]: + sync_connection: typing.Final = typing.cast("Connection", db_session.sync_session.bind) + recorded: typing.Final[list[str]] = [] + + def record(*args: object) -> None: + statement: typing.Final = typing.cast("str", args[2]) + if not statement.startswith(SAVEPOINT_PREFIXES): + recorded.append(statement) + + event.listen(sync_connection, "before_cursor_execute", record) + try: + yield recorded + finally: + event.remove(sync_connection, "before_cursor_execute", record) diff --git a/tests/test_cards.py b/tests/test_cards.py index 6c0d876..321ae3e 100644 --- a/tests/test_cards.py +++ b/tests/test_cards.py @@ -1,13 +1,17 @@ +import datetime from typing import TYPE_CHECKING import pytest +import sqlalchemy as sa from fastapi import status +from app import models from tests import factories if TYPE_CHECKING: from httpx import AsyncClient + from sqlalchemy.ext.asyncio import AsyncSession pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")] @@ -137,6 +141,47 @@ async def test_cards_ordered_by_id_after_update(client: AsyncClient) -> None: assert [x["id"] for x in response.json()["items"]] == [card1.id, card2.id] +async def test_update_cards_sends_one_statement(client: AsyncClient, statements: list[str]) -> None: + deck = await factories.DeckModelFactory.create_async() + card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id) + statements.clear() + + response = await client.put( + f"/api/decks/{deck.id}/cards/", + json=[{"id": card1.id, "front": "card1 updated"}, {"id": card2.id, "front": "card2 updated"}], + ) + assert response.status_code == status.HTTP_200_OK + assert [x["front"] for x in response.json()["items"]] == ["card1 updated", "card2 updated"] + assert len(statements) == 1 + + +async def test_update_cards_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None: + long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) + deck = await factories.DeckModelFactory.create_async() + card = await factories.CardModelFactory.create_async(deck_id=deck.id, updated_at=long_ago) + + response = await client.put(f"/api/decks/{deck.id}/cards/", json=[{"id": card.id, "front": "updated"}]) + assert response.status_code == status.HTTP_200_OK + updated_at = (await db_session.scalars(sa.select(models.Card.updated_at).where(models.Card.id == card.id))).one() + assert updated_at > long_ago + + +async def test_update_cards_with_no_cards(client: AsyncClient) -> None: + deck = await factories.DeckModelFactory.create_async() + + response = await client.put(f"/api/decks/{deck.id}/cards/", json=[]) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"items": []} + + +async def test_update_cards_to_a_duplicate_front(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": card2.id, "front": card1.front}]) + assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT + + async def test_openapi_card_deck_id_required(client: AsyncClient) -> None: response = await client.get("/openapi.json") assert response.status_code == status.HTTP_200_OK diff --git a/tests/test_decks.py b/tests/test_decks.py index 9e1766d..8e171ce 100644 --- a/tests/test_decks.py +++ b/tests/test_decks.py @@ -1,14 +1,18 @@ +import datetime import uuid from typing import TYPE_CHECKING import pytest +import sqlalchemy as sa from fastapi import status +from app import models from tests import factories if TYPE_CHECKING: from httpx import AsyncClient + from sqlalchemy.ext.asyncio import AsyncSession pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")] @@ -72,6 +76,17 @@ async def test_get_one_deck(client: AsyncClient) -> None: assert v == getattr(deck, k) +async def test_get_one_deck_loads_cards_in_one_query(client: AsyncClient, statements: list[str]) -> None: + deck = await factories.DeckModelFactory.create_async() + await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id) + statements.clear() + + response = await client.get(f"/api/decks/{deck.id}/") + assert response.status_code == status.HTTP_200_OK + assert len(response.json()["cards"]) == 2 # noqa: PLR2004 + assert len(statements) == 1 + + @pytest.mark.parametrize( ("name", "description", "status_code"), [ @@ -169,3 +184,23 @@ async def test_post_decks_twice_with_one_idempotency_key_creates_one_deck(client assert second.json() == first.json() response = await client.get("/api/decks/") assert [item["name"] for item in response.json()["items"]] == ["first"] + + +async def test_put_decks_sends_one_statement(client: AsyncClient, statements: list[str]) -> None: + deck = await factories.DeckModelFactory.create_async() + statements.clear() + + response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None}) + assert response.status_code == status.HTTP_200_OK + assert response.json() == {"id": deck.id, "name": "renamed", "description": None} + assert len(statements) == 1 + + +async def test_put_decks_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None: + long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC) + deck = await factories.DeckModelFactory.create_async(updated_at=long_ago) + + response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None}) + assert response.status_code == status.HTTP_200_OK + updated_at = (await db_session.scalars(sa.select(models.Deck.updated_at).where(models.Deck.id == deck.id))).one() + assert updated_at > long_ago