Skip to content

Commit c70a2e3

Browse files
authored
Cut round trips on deck detail, deck update and card upsert (#95)
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.
1 parent 527ccc0 commit c70a2e3

6 files changed

Lines changed: 135 additions & 8 deletions

File tree

‎AGENTS.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ CI (`.github/workflows/main.yml`) runs `ruff format --check`, `ruff check --no-f
4343

4444
`Database`:
4545
- `database_engine` — singleton-ish primary `AsyncEngine` (`create_primary_sa_engine`, `close_sa_engine` finalizer).
46-
- `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.
46+
- `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.
4747
- `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.
4848
- `session` — `Scope.REQUEST`, bound to `dynamic_engine`, finalized by `close_session`.
4949
- `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:
8484
### Database layer
8585

8686
- `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.
87-
- `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`).
87+
- `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`.
8888
- `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.
8989
- `migrations/env.py` swaps the asyncpg driver for the sync `postgresql` driver and uses `app.models.METADATA` as `target_metadata`.
9090

@@ -106,7 +106,7 @@ class ListDecksUseCase:
106106

107107
- 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.
108108
- 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`.
109-
- 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.
109+
- 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.
110110
- Line length is 120.
111111

112112
## Agent skills

‎app/repositories.py‎

Lines changed: 29 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,10 @@
11
import operator
2+
import typing
23
from typing import TYPE_CHECKING
34

4-
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
5+
import sqlalchemy as sa
6+
from advanced_alchemy.exceptions import NotFoundError, wrap_sqlalchemy_exception
7+
from advanced_alchemy.repository import DEFAULT_ERROR_MESSAGE_TEMPLATES, SQLAlchemyAsyncRepository
58
from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
69
from sqlalchemy import orm
710
from sqlalchemy.dialects import postgresql
@@ -33,9 +36,19 @@ async def get_or_create(self, data: schemas.DeckCreate) -> models.Deck:
3336
async def fetch_with_cards(self, deck_id: int) -> models.Deck:
3437
return await self.get_one(
3538
models.Deck.id == deck_id,
36-
load=[orm.selectinload(models.Deck.cards)],
39+
load=[orm.joinedload(models.Deck.cards)],
40+
uniquify=True,
3741
)
3842

43+
async def update_deck(self, deck_id: int, data: schemas.DeckUpdate) -> models.Deck:
44+
statement: typing.Final = (
45+
sa.update(models.Deck).where(models.Deck.id == deck_id).values(**data.model_dump()).returning(models.Deck)
46+
)
47+
deck: typing.Final = (await self.repository.session.scalars(statement)).one_or_none()
48+
if deck is None:
49+
raise NotFoundError
50+
return deck
51+
3952

4053
class CardsRepository(SQLAlchemyAsyncRepositoryService[models.Card]):
4154
class BaseRepository(SQLAlchemyAsyncRepository[models.Card]):
@@ -50,7 +63,19 @@ async def add_cards(self, deck_id: int, cards: list[schemas.CardCreate]) -> Sequ
5063
return await self.create_many([models.Card(**card.model_dump(), deck_id=deck_id) for card in cards])
5164

5265
async def upsert_cards(self, deck_id: int, cards: list[schemas.CardUpdate]) -> Sequence[models.Card]:
53-
upserted = await self.upsert_many(
54-
[models.Card(**card.model_dump(), deck_id=deck_id) for card in cards],
66+
if not cards:
67+
return []
68+
statement = postgresql.insert(models.Card).values([{**card.model_dump(), "deck_id": deck_id} for card in cards])
69+
statement = statement.on_conflict_do_update(
70+
index_elements=[models.Card.id],
71+
set_={
72+
"front": statement.excluded.front,
73+
"back": statement.excluded.back,
74+
"hint": statement.excluded.hint,
75+
"deck_id": statement.excluded.deck_id,
76+
"updated_at": sa.func.now(),
77+
},
5578
)
79+
with wrap_sqlalchemy_exception(error_messages=DEFAULT_ERROR_MESSAGE_TEMPLATES, dialect_name="postgresql"):
80+
upserted: typing.Final = (await self.repository.session.scalars(statement.returning(models.Card))).all()
5681
return sorted(upserted, key=operator.attrgetter("id"))

‎app/use_cases/update_deck.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,6 @@ class UpdateDeckUseCase:
1515
@postgres_retry
1616
async def __call__(self, *, deck_id: int, data: schemas.DeckUpdate) -> models.Deck:
1717
async with self.transaction:
18-
deck: typing.Final = await self.decks_repository.update(data=data.model_dump(), item_id=deck_id)
18+
deck: typing.Final = await self.decks_repository.update_deck(deck_id, data)
1919
await self.transaction.commit()
2020
return deck

‎tests/conftest.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
from asgi_lifespan import LifespanManager
66
from httpx import ASGITransport, AsyncClient
77
from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
8+
from sqlalchemy import event
89

910
from app import ioc
1011
from app.application import build_app
@@ -14,9 +15,13 @@
1415
if typing.TYPE_CHECKING:
1516
import fastapi
1617
import modern_di
18+
from sqlalchemy.engine import Connection
1719
from sqlalchemy.ext.asyncio import AsyncSession
1820

1921

22+
SAVEPOINT_PREFIXES: typing.Final = ("SAVEPOINT", "RELEASE SAVEPOINT", "ROLLBACK TO SAVEPOINT")
23+
24+
2025
@pytest.fixture
2126
async def app() -> typing.AsyncIterator[fastapi.FastAPI]:
2227
app_ = build_app()
@@ -68,3 +73,20 @@ async def set_async_session_in_base_sqlalchemy_factory(
6873
yield
6974
finally:
7075
SQLAlchemyFactory.__async_session__ = None
76+
77+
78+
@pytest.fixture
79+
def statements(db_session: AsyncSession) -> typing.Iterator[list[str]]:
80+
sync_connection: typing.Final = typing.cast("Connection", db_session.sync_session.bind)
81+
recorded: typing.Final[list[str]] = []
82+
83+
def record(*args: object) -> None:
84+
statement: typing.Final = typing.cast("str", args[2])
85+
if not statement.startswith(SAVEPOINT_PREFIXES):
86+
recorded.append(statement)
87+
88+
event.listen(sync_connection, "before_cursor_execute", record)
89+
try:
90+
yield recorded
91+
finally:
92+
event.remove(sync_connection, "before_cursor_execute", record)

‎tests/test_cards.py‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,17 @@
1+
import datetime
12
from typing import TYPE_CHECKING
23

34
import pytest
5+
import sqlalchemy as sa
46
from fastapi import status
57

8+
from app import models
69
from tests import factories
710

811

912
if TYPE_CHECKING:
1013
from httpx import AsyncClient
14+
from sqlalchemy.ext.asyncio import AsyncSession
1115

1216

1317
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:
137141
assert [x["id"] for x in response.json()["items"]] == [card1.id, card2.id]
138142

139143

144+
async def test_update_cards_sends_one_statement(client: AsyncClient, statements: list[str]) -> None:
145+
deck = await factories.DeckModelFactory.create_async()
146+
card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
147+
statements.clear()
148+
149+
response = await client.put(
150+
f"/api/decks/{deck.id}/cards/",
151+
json=[{"id": card1.id, "front": "card1 updated"}, {"id": card2.id, "front": "card2 updated"}],
152+
)
153+
assert response.status_code == status.HTTP_200_OK
154+
assert [x["front"] for x in response.json()["items"]] == ["card1 updated", "card2 updated"]
155+
assert len(statements) == 1
156+
157+
158+
async def test_update_cards_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None:
159+
long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
160+
deck = await factories.DeckModelFactory.create_async()
161+
card = await factories.CardModelFactory.create_async(deck_id=deck.id, updated_at=long_ago)
162+
163+
response = await client.put(f"/api/decks/{deck.id}/cards/", json=[{"id": card.id, "front": "updated"}])
164+
assert response.status_code == status.HTTP_200_OK
165+
updated_at = (await db_session.scalars(sa.select(models.Card.updated_at).where(models.Card.id == card.id))).one()
166+
assert updated_at > long_ago
167+
168+
169+
async def test_update_cards_with_no_cards(client: AsyncClient) -> None:
170+
deck = await factories.DeckModelFactory.create_async()
171+
172+
response = await client.put(f"/api/decks/{deck.id}/cards/", json=[])
173+
assert response.status_code == status.HTTP_200_OK
174+
assert response.json() == {"items": []}
175+
176+
177+
async def test_update_cards_to_a_duplicate_front(client: AsyncClient) -> None:
178+
deck = await factories.DeckModelFactory.create_async()
179+
card1, card2 = await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
180+
181+
response = await client.put(f"/api/decks/{deck.id}/cards/", json=[{"id": card2.id, "front": card1.front}])
182+
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
183+
184+
140185
async def test_openapi_card_deck_id_required(client: AsyncClient) -> None:
141186
response = await client.get("/openapi.json")
142187
assert response.status_code == status.HTTP_200_OK

‎tests/test_decks.py‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,18 @@
1+
import datetime
12
import uuid
23
from typing import TYPE_CHECKING
34

45
import pytest
6+
import sqlalchemy as sa
57
from fastapi import status
68

9+
from app import models
710
from tests import factories
811

912

1013
if TYPE_CHECKING:
1114
from httpx import AsyncClient
15+
from sqlalchemy.ext.asyncio import AsyncSession
1216

1317

1418
pytestmark = [pytest.mark.usefixtures("set_async_session_in_base_sqlalchemy_factory")]
@@ -72,6 +76,17 @@ async def test_get_one_deck(client: AsyncClient) -> None:
7276
assert v == getattr(deck, k)
7377

7478

79+
async def test_get_one_deck_loads_cards_in_one_query(client: AsyncClient, statements: list[str]) -> None:
80+
deck = await factories.DeckModelFactory.create_async()
81+
await factories.CardModelFactory.create_batch_async(size=2, deck_id=deck.id)
82+
statements.clear()
83+
84+
response = await client.get(f"/api/decks/{deck.id}/")
85+
assert response.status_code == status.HTTP_200_OK
86+
assert len(response.json()["cards"]) == 2 # noqa: PLR2004
87+
assert len(statements) == 1
88+
89+
7590
@pytest.mark.parametrize(
7691
("name", "description", "status_code"),
7792
[
@@ -169,3 +184,23 @@ async def test_post_decks_twice_with_one_idempotency_key_creates_one_deck(client
169184
assert second.json() == first.json()
170185
response = await client.get("/api/decks/")
171186
assert [item["name"] for item in response.json()["items"]] == ["first"]
187+
188+
189+
async def test_put_decks_sends_one_statement(client: AsyncClient, statements: list[str]) -> None:
190+
deck = await factories.DeckModelFactory.create_async()
191+
statements.clear()
192+
193+
response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None})
194+
assert response.status_code == status.HTTP_200_OK
195+
assert response.json() == {"id": deck.id, "name": "renamed", "description": None}
196+
assert len(statements) == 1
197+
198+
199+
async def test_put_decks_moves_updated_at(client: AsyncClient, db_session: AsyncSession) -> None:
200+
long_ago = datetime.datetime(2000, 1, 1, tzinfo=datetime.UTC)
201+
deck = await factories.DeckModelFactory.create_async(updated_at=long_ago)
202+
203+
response = await client.put(f"/api/decks/{deck.id}/", json={"name": "renamed", "description": None})
204+
assert response.status_code == status.HTTP_200_OK
205+
updated_at = (await db_session.scalars(sa.select(models.Deck.updated_at).where(models.Deck.id == deck.id))).one()
206+
assert updated_at > long_ago

0 commit comments

Comments
 (0)