From 7d62b7d16a94f8c7f063e20406636b12ce81a099 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 13:29:15 +0300 Subject: [PATCH] Cut round trips on deck detail, deck update and card upsert Deck detail loads cards with joinedload (one query instead of two). Deck update is one UPDATE ... RETURNING instead of advanced-alchemy's update, which read the row twice and refreshed it. Card upsert is one INSERT ... ON CONFLICT (id) DO UPDATE ... RETURNING instead of upsert_many, which selected the rows before and after writing them. --- AGENTS.md | 6 ++--- app/repositories.py | 33 ++++++++++++++++++++++---- app/use_cases/update_deck.py | 2 +- tests/conftest.py | 22 ++++++++++++++++++ tests/test_cards.py | 45 ++++++++++++++++++++++++++++++++++++ tests/test_decks.py | 35 ++++++++++++++++++++++++++++ 6 files changed, 135 insertions(+), 8 deletions(-) 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