$npx -y skills add ocbunknown/fastapi-claude-template --skill usecaseUse when creating or editing a use case under src/application/v1/usecases/. Enforces the RequestBus + UseCase pattern (Request + UseCase in one file, @dataclass(slots=True), returns Result type) and the DBGateway transaction rules — never open two sessions in one use case, never
| 1 | # Writing use cases (`src/application/v1/usecases/<domain>/`) |
| 2 | |
| 3 | A use case is the **transport-agnostic unit of business logic**. Endpoints, consumers, and scheduled tasks all call `request_bus.send(Request(...))` and get back a `Result`. Use cases never know who called them. |
| 4 | |
| 5 | ## One file, two classes |
| 6 | |
| 7 | Every use case file defines **exactly one** `Request` class and **exactly one** `UseCase` class. Keep them in the same file — they are a pair. |
| 8 | |
| 9 | ```python |
| 10 | # src/application/v1/usecases/widget/create.py |
| 11 | from dataclasses import dataclass |
| 12 | from typing import Annotated |
| 13 | |
| 14 | import uuid_utils.compat as uuid |
| 15 | from pydantic import Field |
| 16 | |
| 17 | from src.application.common.interfaces.usecase import UseCase |
| 18 | from src.application.common.request import Request |
| 19 | from src.application.v1.results import WidgetResult |
| 20 | from src.database.psql import DBGateway |
| 21 | |
| 22 | |
| 23 | class CreateWidgetRequest(Request): |
| 24 | name: Annotated[str, Field(max_length=64)] |
| 25 | owner_uuid: uuid.UUID |
| 26 | |
| 27 | |
| 28 | @dataclass(slots=True) |
| 29 | class CreateWidgetUseCase(UseCase[CreateWidgetRequest, WidgetResult]): |
| 30 | database: DBGateway |
| 31 | |
| 32 | async def __call__(self, request: CreateWidgetRequest) -> WidgetResult: |
| 33 | async with self.database: |
| 34 | widget = ( |
| 35 | await self.database.widget.create( |
| 36 | name=request.name, |
| 37 | owner_uuid=request.owner_uuid, |
| 38 | ) |
| 39 | ).result() |
| 40 | return WidgetResult(**widget.as_dict()) |
| 41 | ``` |
| 42 | |
| 43 | ## Invariants |
| 44 | |
| 45 | 1. **`Request`** inherits `src.application.common.request.Request` (Pydantic, `frozen=True`). Validation via `Annotated[..., Field(...)]`. Do not reuse HTTP `Contract` classes here — the Request describes the use-case input, not the HTTP input. |
| 46 | 2. **`UseCase`** inherits `UseCase[RequestClass, ReturnType]`. The base class auto-applies `@dataclass(slots=True)` — add your own `@dataclass(slots=True)` on the subclass too for mypy/readability. |
| 47 | 3. **Dependencies are dataclass fields**: `database: DBGateway`, `hasher: Hasher`, `cache: StrCache`, `services: ServiceGateway`, etc. Injected by Dishka via the request bus — do not instantiate manually. |
| 48 | 4. **Returns a `Result` subclass** from `src/application/v1/results/` — never a `Contract` from `presentation/`, never a bare ORM model, never a dict. |
| 49 | 5. **Register** the use case in `src/application/v1/usecases/__init__.py::setup_use_cases()`: |
| 50 | ```python |
| 51 | request_bus.register(widget.CreateWidgetRequest, widget.CreateWidgetUseCase) |
| 52 | ``` |
| 53 | 6. **Re-export** from `src/application/v1/usecases/widget/__init__.py` so the presentation/consumers layer can import `from src.application.v1.usecases.widget import CreateWidgetRequest`. |
| 54 | |
| 55 | ## DBGateway — the transaction rule |
| 56 | |
| 57 | This is the **single most common place people break production**. Read carefully. |
| 58 | |
| 59 | `DBGateway` owns one session per request. It exposes two context managers. **You pick exactly one per use case — never both, never twice.** |
| 60 | |
| 61 | ### Pattern 1 — read-only |
| 62 | |
| 63 | ```python |
| 64 | async def __call__(self, request: SelectWidgetRequest) -> WidgetResult: |
| 65 | async with self.database.manager.session: |
| 66 | widget = ( |
| 67 | await self.database.widget.select(*request.loads, widget_uuid=request.widget_uuid) |
| 68 | ).result() |
| 69 | return WidgetResult(**widget.as_dict()) |
| 70 | ``` |
| 71 | |
| 72 | - Use for pure reads: one or many `select` / `exists` / `count` calls, no mutations. |
| 73 | - No transaction is opened — `session.begin()` is never called. |
| 74 | - Build the Result via `WidgetResult(**widget.as_dict())` — see "Mapping ORM → Result" below for why `.model_validate(widget)` is forbidden. |
| 75 | - Keep construction **inside** the `async with` block by convention — it localizes session-touching code and keeps the pattern consistent even when the use case later grows logic that genuinely needs the open session. |
| 76 | |
| 77 | ### Pattern 2 — write |
| 78 | |
| 79 | ```python |
| 80 | async def __call__(self, request: CreateWidgetRequest) -> WidgetResult: |
| 81 | async with self.database: |
| 82 | widget = ( |
| 83 | await self.database.widget.create( |
| 84 | name=request.name, |
| 85 | owner_uuid=request.owner_uuid, |
| 86 | ) |
| 87 | ).result() |
| 88 | return WidgetResult(**widget.as_dict()) |
| 89 | ``` |
| 90 | |
| 91 | - Use for any mutation (`create` / `update` / `delete` / `upsert`). |
| 92 | - `async with self.database:` calls `session.begin()` → autocommit on success, autorollback on exception. |
| 93 | - Every `self.database.X` call and every `Result` construction belongs **inside** this block. |
| 94 | |
| 95 | ### Pattern 3 — read then write (single transaction) |
| 96 | |
| 97 | When a write needs data that was just read, put **both** in one `async with self.database:` block: |
| 98 | |
| 99 | ```python |
| 100 | async def __call__(self, request: Confi |