Development Conventions¶

Code Style¶

Python¶

• Formatter: Black (line length: 88)
• Linter: Ruff
• Type checker: mypy (strict mode)
• Import sorting: isort

# Good example
from uuid import UUID

from pydantic import BaseModel, ConfigDict, Field


class RecipeId(BaseModel):
    """Recipe identifier value object."""

    model_config = ConfigDict(frozen=True)

    value: UUID = Field(..., description="Unique recipe identifier")

Type Hints¶

• All public functions must have type hints
• Use | for unions (Python 3.10+)
• Use list[T], dict[K, V] instead of List, Dict

# Good
def get_recipe(recipe_id: RecipeId) -> Recipe | None:
    ...

# Bad
def get_recipe(recipe_id):  # No type hints
    ...

Naming Conventions¶

Module Layout (canonical)¶

Every module follows the same skeleton. Default to folder-per-layer-piece even when there is only one entity — uniformity across the seven modules beats local minimalism.

src/modules/<m>/
├── domain/
│   ├── entities/
│   │   ├── __init__.py
│   │   ├── recipe.py           # one entity per file
│   │   └── bom_line.py
│   ├── value_objects/
│   │   ├── __init__.py
│   │   ├── recipe_status.py
│   │   └── yield_quantity.py
│   ├── repositories/           # ABCs only
│   │   ├── recipe_repository.py
│   │   └── bom_line_repository.py
│   ├── ports/                  # outbound interfaces to other modules
│   │   ├── warehouse_stock_port.py
│   │   └── crm_orderable_port.py
│   ├── events.py               # PascalCase DomainEvent subclasses pinned to this module
│   └── exceptions.py
├── application/
│   ├── commands/
│   │   ├── create_recipe.py    # CreateRecipeCommand + CreateRecipeResult + CreateRecipeHandler
│   │   └── publish_recipe_version.py
│   ├── queries/
│   │   ├── get_recipe.py       # GetRecipeQuery + RecipeReadModel + GetRecipeHandler
│   │   └── list_recipes.py
│   └── dtos.py                 # ONLY types reused by multiple handlers; per-use-case results live with their handler
├── infrastructure/
│   ├── models/                 # one SQLAlchemy model per file
│   │   ├── __init__.py         # re-exports every model so it lands on Base.metadata
│   │   ├── recipe.py
│   │   └── bom_line.py
│   ├── repositories/           # one SQLAlchemy repo impl per file
│   │   ├── sqlalchemy_recipe_repository.py
│   │   └── sqlalchemy_bom_line_repository.py
│   └── event_publisher.py
└── api/
    ├── router.py               # composes routers/ → exposes one <module>_router
    ├── routers/                # one APIRouter per entity / use-case grouping
    │   ├── recipes.py
    │   └── recipe_versions.py
    ├── dependencies.py         # Annotated dependency aliases for this module
    └── schemas/
        ├── __init__.py
        ├── shared.py           # RecipeRead, IdResponse, PageMeta — used by many routes
        ├── create_recipe.py    # CreateRecipeRequest / CreateRecipeResponse
        └── publish_recipe_version.py

Naming: snake_case files; class name in PascalCase. The file entities/recipe.py contains class Recipe. The file commands/create_recipe.py contains both CreateRecipeCommand and CreateRecipeHandler. The file schemas/create_recipe.py contains both CreateRecipeRequest and CreateRecipeResponse.

CQS Pattern (Command Query Separation)¶

All application-layer operations are either a Command (writes state, has side effects) or a Query (reads state, no side effects). The naming makes this immediately visible.

Command Handler¶

Commands change state. They save to DB and emit domain events. Named {Action}{Noun}Command + {Action}{Noun}Handler.

# application/commands/create_recipe.py
from dataclasses import dataclass
from uuid import UUID

from modules.recipe.domain.entities.recipe import Recipe
from modules.recipe.domain.exceptions import RecipeCodeAlreadyExistsError
from modules.recipe.domain.repositories.recipe_repository import RecipeRepository
from src.shared.observability import get_logger, traced_handler

logger = get_logger(__name__)


@dataclass
class CreateRecipeCommand:
    name: str
    code: str
    category: str
    tenant_id: UUID
    created_by: UUID


@dataclass
class CreateRecipeResult:
    recipe_id: UUID


class CreateRecipeHandler:
    def __init__(self, recipe_repo: RecipeRepository) -> None:
        self._recipe_repo = recipe_repo

    @traced_handler
    async def handle(self, command: CreateRecipeCommand) -> CreateRecipeResult:
        # load → validate → call domain → save → emit → return
        existing = await self._recipe_repo.get_by_code(command.code, command.tenant_id)
        if existing:
            raise RecipeCodeAlreadyExistsError(command.code)

        recipe = Recipe.create(
            name=command.name,
            code=command.code,
            category=command.category,
            tenant_id=command.tenant_id,
            created_by=command.created_by,
        )
        await self._recipe_repo.save(recipe)
        logger.info("recipe_created", event="recipe_created", entity_id=str(recipe.id))
        return CreateRecipeResult(recipe_id=recipe.id)

@traced_handler opens a span named {module}.{handler_class} and binds module / command keys to structlog.contextvars for the duration of the call. Use logger = get_logger(__name__) at the top of every module so each log record carries its fully-qualified module path under the logger key — searchable in SigNoz. Query handlers use the same decorator; on the happy path a Query Handler typically logs nothing — the HTTP span already records duration and status.

File-layout rule: the Command, Result, and Handler for one use case live together in application/commands/{use_case}.py (and the equivalent for queries under application/queries/). application/dtos.py is reserved for application-layer types that are genuinely shared across two or more handlers — for example a read model returned by both GetItem and ListItems. Per-use-case Result / ReadModel dataclasses do not go in dtos.py; keep them next to the handler that produces them so the entire use case is one file to read.

Query Handler¶

Queries only read. No saves, no events, no domain method calls that mutate state. Named {Get/List}{Noun}Query + {Get/List}{Noun}Handler.

# application/queries/get_recipe.py
from dataclasses import dataclass
from uuid import UUID

from modules.recipe.domain.exceptions import RecipeNotFoundError
from modules.recipe.domain.repositories.recipe_repository import RecipeRepository
from src.shared.observability import traced_handler


@dataclass
class GetRecipeQuery:
    recipe_id: UUID
    tenant_id: UUID


@dataclass
class RecipeReadModel:
    recipe_id: UUID
    name: str
    code: str
    status: str
    current_version: int


class GetRecipeHandler:
    def __init__(self, recipe_repo: RecipeRepository) -> None:
        self._recipe_repo = recipe_repo

    @traced_handler
    async def handle(self, query: GetRecipeQuery) -> RecipeReadModel:
        recipe = await self._recipe_repo.get_by_id(query.recipe_id)
        if recipe is None:
            raise RecipeNotFoundError(query.recipe_id)
        return RecipeReadModel(
            recipe_id=recipe.id,
            name=recipe.name,
            code=recipe.code,
            status=recipe.status.value,
            current_version=recipe.current_version,
        )

Domain-Driven Design¶

Aggregate Rules¶

1. One repository per aggregate root
2. Transaction boundary = aggregate boundary
3. Keep aggregates small - only include what's needed for consistency
4. Reference other aggregates by ID, not by object reference

# Good - reference by ID
class ProductionOrder:
    recipe_id: RecipeId  # Not Recipe object
    recipe_version: int

# Bad - holding reference
class ProductionOrder:
    recipe: Recipe  # Don't do this

Value Objects¶

• Immutable - use frozen=True in Pydantic
• Equality by value - two VOs with same values are equal
• No identity - only value matters

class Quantity(BaseModel):
    model_config = ConfigDict(frozen=True)

    value: Decimal
    unit: UnitOfMeasure

    def __add__(self, other: "Quantity") -> "Quantity":
        if self.unit != other.unit:
            raise UnitMismatchError
        return Quantity(value=self.value + other.value, unit=self.unit)

Domain Events¶

• Named in past tense: RecipeCreated, OrderCompleted
• Contain all data needed by consumers
• Include tenant_id for routing

class RecipeVersionPublished(DomainEvent):
    event_type: Literal["RecipeVersionPublished"] = "RecipeVersionPublished"
    recipe_id: UUID
    version: int
    author_id: UUID
    published_at: datetime

Guard Placement — Where Invariants Live¶

Every command handler does some checking before it mutates state. Where a given check belongs is not a style decision — putting it in the wrong layer either weakens the model (entity stops being a source of truth) or duplicates logic that drifts over time.

Three categories, three layers:

Handlers MUST NOT duplicate pure state-machine invariants. The entity already enforces them; a duplicate check in the handler creates two sources of truth that will drift. If you find yourself writing if item.status is X: raise Y in a handler, and Item.method() already raises Y for that case, delete the handler-level check and let the domain raise.

The one exception — error-precedence short-circuit. A handler MAY pre-check a state-machine condition if and only if BOTH:

1. There are subsequent expensive operations (cross-module port calls, large reads) that would run before the domain method, and
2. The error semantics require the state-machine error to take precedence over those other failures.

When this exception applies, document it with an inline comment so the next reader knows why the duplication is intentional.

# application/commands/deactivate_item.py — EXCEPTION applies
async def handle(self, command: DeactivateItemCommand) -> DeactivateItemResult:
    item = await self._item_repo.get_by_id(command.item_id, command.tenant_id)
    if item is None:
        raise ItemNotFoundError(...)

    if item.is_system:                          # actor-aware authorization — handler
        raise SystemItemIsReadOnlyError(...)

    # State-machine pre-check — duplicates Item.deactivate's invariant on purpose:
    # without it, an already-INACTIVE item that also has stock would surface as
    # 409 ITEM_HAS_STOCK_REMAINING instead of the correct 400 INVALID_ITEM_STATE_TRANSITION.
    if item.status is ItemStatus.INACTIVE:
        raise InvalidItemStateTransitionError(...)

    stock = await self._warehouse_port.get_stock_quantity(...)   # expensive cross-module call
    if stock > 0:
        raise ItemHasStockRemainingError(...)
    # ... more cross-module checks ...

    item.deactivate(deactivated_by=command.deactivated_by)
    await self._item_repo.save(item)
    await self._event_publisher.publish(_build_item_deactivated(item))
    return DeactivateItemResult(item_id=item.id)

# application/commands/reactivate_item.py — EXCEPTION does NOT apply
async def handle(self, command: ReactivateItemCommand) -> ReactivateItemResult:
    item = await self._item_repo.get_by_id(command.item_id, command.tenant_id)
    if item is None:
        raise ItemNotFoundError(...)

    # No cross-module checks follow → no precedence problem → no handler-level
    # state-machine guard. Item.reactivate() raises InvalidItemStateTransitionError
    # for already-ACTIVE items; that's the single source of truth.
    item.reactivate(reactivated_by=command.reactivated_by)
    await self._item_repo.save(item)
    await self._event_publisher.publish(_build_item_reactivated(item))
    return ReactivateItemResult(item_id=item.id)

Test placement follows guard placement:

• State-machine invariants → domain unit test (tests/unit/{module}/domain/test_{entity}.py). The handler test does not re-assert them — that would test the same thing twice and couple handler tests to domain internals.
• Actor-aware authorization → handler unit test (tests/unit/{module}/application/test_{command}.py).
• Cross-aggregate guards → handler unit test (with fake ports / repos).
• The error-precedence exception, when it applies → one handler unit test asserting the precedence (i.e. configure two failures at once and assert the state-machine one wins). Don't repeat the domain's own happy/sad paths — just the ordering.

Repository Pattern¶

• Interface in domain layer (Protocol/class)
• Implementation in infrastructure layer
• Only for aggregate roots

# domain/repositories/recipe_repository.py
class RecipeRepository(Protocol):
    async def get_by_id(self, recipe_id: RecipeId) -> Recipe | None: ...
    async def save(self, recipe: Recipe) -> None: ...


# infrastructure/persistence/repositories/sqlalchemy_recipe_repository.py
class SqlAlchemyRecipeRepository(RecipeRepository):
    def __init__(self, session: AsyncSession):
        self._session = session

    async def get_by_id(self, recipe_id: RecipeId) -> Recipe | None:
        model = await self._session.get(RecipeModel, recipe_id.value)
        return RecipeConverter.to_domain(model) if model else None

Testing¶

The authoritative source for test layout, factory patterns, fixture reference, parametrization rules, and per-layer coverage targets is docs/testing/testing-conventions.md. Read it in full before writing any test.

Highlights for cross-cutting decisions made elsewhere in this doc:

• Tests live in the same PR as the code they cover — never in a follow-up.
• Four test layers mirror the four DDD layers (domain unit, application unit, infrastructure integration, API integration). Each layer uses the lightest infrastructure that still proves the behaviour — domain tests touch no DB, application tests use in-memory repos, infrastructure tests use a real Postgres, API tests go end-to-end through FastAPI.
• Parametrize related variants (@pytest.mark.parametrize with ids=[...]) when the assertion logic is identical and only the inputs differ — see testing-conventions.md § Parametrize Related Test Variants. Apply the rule of three: when you write the third near-identical test, collapse them.
• Class-per-behaviour test organisation: one Test{Command} class per command/query handler, one Test{Endpoint} class per route. Class-scoped fixtures provide the SUT.
• Outbox assertion is mandatory at the API integration layer for every state-mutating endpoint — see § Transactional Outbox below for the contract.

Per-layer coverage targets, factory style, fixture reference, and worked examples for each layer all live in the testing-conventions doc.

Git Conventions¶

Branch Naming¶

feat/REC-123-multi-level-bom
fix/WH-456-fefo-sorting
docs/api-documentation
refactor/PROD-789-order-workflow
test/REC-456-recipe-tests

Commit Messages (Conventional Commits)¶

<type>(<scope>): <description>

[optional body]

[optional footer]

Types: - feat - New feature - fix - Bug fix - docs - Documentation - refactor - Code refactoring - test - Tests - chore - Maintenance

Examples:

feat(recipe): add multi-level BOM support

- Add RecipeIngredient with semi-finished flag
- Implement BOM traversal for cost calculation
- Add max depth validation (5 levels)

Closes REC-123

fix(warehouse): correct FEFO sorting for expired batches

Expiration date was not considered when lots had same received date.

Fixes WH-456

Pull Requests¶

Title: Same format as commit message

Template:

## Summary
Brief description of changes.

## Changes
- Bullet list of key changes

## Testing
- [ ] Unit tests added/updated
- [ ] Integration tests pass
- [ ] Manual testing done

## Related
- Closes #123

API Conventions¶

Schemas vs Commands vs Queries¶

Three distinct types — each lives at a different layer:

The API route maps between layers:

# api/router.py
@router.post("/recipes", status_code=201)
async def create_recipe(
    body: CreateRecipeRequest,                    # ← Pydantic schema
    handler: CreateRecipeHandler = Depends(...),
    current_user: CurrentUser = Depends(get_current_user),
) -> CreateRecipeResponse:                        # ← Pydantic schema
    result = await handler.handle(
        CreateRecipeCommand(                      # ← Command dataclass
            name=body.name,
            code=body.code,
            category=body.category,
            tenant_id=current_user.tenant_id,
            created_by=current_user.user_id,
        )
    )
    return CreateRecipeResponse(recipe_id=result.recipe_id)

Why not just use Pydantic everywhere? Commands and Queries must be testable without FastAPI. Call handler.handle(CreateRecipeCommand(...)) directly in integration tests — no HTTP client, no Pydantic validation overhead. The schema layer stays strictly at the HTTP boundary.

Naming rule: If a class name ends in Command, Query, or Result/ReadModel — it is a plain dataclass in the application layer. If it ends in Request or Response — it is a Pydantic model in the API layer.

Schema field typing — enum-domain fields must use the enum class (mandatory)¶

Any Pydantic field — on either a request or response schema — whose semantic domain is an enum must be typed with that enum class. Bare str is banned for those fields. This applies to every enum we expose: domain StrEnums (ItemType, ItemStatus, StorageUnit, …) and hand-written Literal[...] unions (e.g. sort tokens, the tri-state status filter).

# ❌ BREAKS FE codegen — OpenAPI emits "type": "string"
class ItemDetailResponse(BaseModel):
    item_type: str
    storage_unit: str
    status: str

# ✅ CORRECT — OpenAPI emits $ref to the enum component
class ItemDetailResponse(BaseModel):
    item_type: ItemType
    storage_unit: StorageUnit
    status: ItemStatus

Why this is a hard rule. Both FE codegen passes depend on the enum being visible in the spec:

• Orval → TS types. Field typed str becomes string; field typed with the enum becomes the discriminated union ('active' | 'inactive'). Every FE component that branches on item.status === 'active' only narrows correctly when the field is the union — string never narrows and the branch silently falls through to the fallback render.
• Orval → MSW faker mocks. Field typed str becomes faker.string.alpha({length:{min:10,max:20}}) — the dev mock returns nonsense like "qweRTYuiopASD" that no === branch ever matches. Enum-typed fields become faker.helpers.arrayElement([...]) and the FE renders the same code paths in pnpm dev:msw as in production.

The runtime BE values are already constrained to the enum — this rule is about the advertised type, not the actual type.

status gotcha — domain enum on responses, tri-state filter on query params. A list endpoint's query model uses ItemStatusFilter (active | inactive | all) for ?status= because "all" is a valid request. The matching response schemas use the domain enum ItemStatus (active | inactive) — the response can never return all. The Literal["active","inactive","all"] filter rule lives in § Status filter — tri-state enum further down; the response-side rule is here.

Enforcement. tests/unit/shared/api/test_openapi_enum_consistency.py walks the generated OpenAPI spec and fails if any property name is typed as bare string in one schema while being $ref'd to an enum component in another. New endpoints that violate the rule break CI immediately — no separate review check needed.

Router Composition¶

Three tiers, top-down:

1. Entity / use-case routers live in src/modules/<m>/api/routers/. Each file declares one APIRouter with a prefix matching its resource (e.g. items_router = APIRouter(prefix="/items", tags=["item-catalog"])).
2. Module router in src/modules/<m>/api/router.py composes those into a single <module>_router = APIRouter(), including each entity router with <module>_router.include_router(items_router).
3. Global router in src/shared/api/router.py declares api_router = APIRouter(prefix="/api/v1") and include_routers every module router.

src/main.py mounts exactly two routers: health_router and the global api_router. Module-level changes never touch main.py.

Dependency Injection¶

Every dependency that appears in more than one route must be exposed as an Annotated[T, Depends(...)] alias. This keeps route signatures readable and ensures the dependency is wired the same way everywhere.

Cross-module aliases live in src/shared/api/deps.py:

from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from src.shared.auth.current_user import CurrentUser, get_current_user
from src.shared.db.session import get_session

DbSession = Annotated[AsyncSession, Depends(get_session)]
CurrentUserDep = Annotated[CurrentUser, Depends(get_current_user)]

Module aliases live in src/modules/<m>/api/dependencies.py:

from typing import Annotated
from fastapi import Depends

ItemRepoDep = Annotated[ItemRepository, Depends(get_item_repository)]
ItemGroupRepoDep = Annotated[ItemGroupRepository, Depends(get_item_group_repository)]

Usage in a route:

@router.post("", status_code=201)
async def create_item(
    body: CreateItemRequest,
    repo: ItemRepoDep,
    user: CurrentUserDep,
) -> CreateItemResponse:
    ...

Role-gated endpoints use the permission-based guard from src.shared.auth — Depends(require_permission(Permission.X)) — rather than pre-built combination aliases. See § Authorization below for the full pattern.

URL Structure¶

/api/v1/{resource}
/api/v1/{resource}/{id}
/api/v1/{resource}/{id}/sub-resource

HTTP Methods¶

Response Codes¶

Pagination¶

Offset-based, 1-indexed page numbers, with a stable snapshot_at returned in the response envelope to defend against duplicates when rows are inserted between page fetches. See § List Endpoints below for the full design (pagination, filtering, sorting, search, stats banner, FE coordination).

GET /api/v1/recipes?page=1&page_size=20

Response:
{
  "items": [...],
  "total": 100,
  "page": 1,
  "page_size": 20,
  "pages": 5,
  "snapshot_at": "2026-05-16T12:34:56Z"
}

Error Response¶

The runtime envelope every non-2xx response carries. Routes declare which of these envelopes are possible via the errors_for() helper — see § Error Response Declarations below.

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Validation failed",
    "details": [
      {"field": "name", "message": "Name is required"}
    ]
  }
}

Error Response Declarations¶

### Error Response above describes the runtime envelope. This section describes the OpenAPI contract — how each route advertises its possible error responses so the FE (and any other consumer) can read them from the spec instead of grepping domain/exceptions.py.

Rule: every route declares its possible errors via errors_for(...) from src.shared.api.error_responses:

from src.shared.api.error_responses import errors_for

@items_router.patch(
    "/{item_id}",
    status_code=200,
    response_model=EditItemResponse,
    responses=errors_for(
        domain=[EANAlreadyInUseError, SystemItemIsReadOnlyError, ShelfLifeDaysNotApplicableError],
        not_found=ItemNotFoundError,
        mutation=True,
        optimistic=True,
        patch=True,
        permission=Permission.ITEM_UPDATE,
    ),
)

errors_for() builds a FastAPI responses= dict from three sources:

• Class-universal entries (always added). Every route can return 422 VALIDATION_ERROR and 500 REPOSITORY_ERROR, so the helper emits them unconditionally.
• Flag-implied entries (added when the flag is set). Each flag corresponds to a piece of shared machinery that fires the same exception across every route that opts in:

• Domain entries (domain=[...]) — the per-route list of DomainError subclasses the handler can raise. Each class is added at its own http_status.

Grouping by status. Multiple exception classes that share the same HTTP status (e.g. several 409s) collapse into a single OpenAPI response entry whose examples field carries one named example per exception (keyed by code). Swagger UI shows the full list of possible codes inline.

Routes with no domain errors still call errors_for() so the class-universal tier and the flag-implied entries are wired correctly — "drop the kwarg" is never a valid shortcut.

Identity-style endpoints opt out of 401. Login / token-refresh / register endpoints don't go through any auth dependency, so they cannot return the generic UNAUTHORIZED. They pass authenticated=False; any 401 they emit (e.g. bad credentials) is a domain error listed in domain=[...].

Backfill obligation. Every module's first PR — and every new route after that — must use errors_for(). The OpenAPI spec is the single FE-facing contract; routes that don't declare their errors silently break planning for downstream consumers.

The FE-side parser, i18n catalogue, and handleApiError helper that consume this contract are tracked separately (see grinsystem-front/docs/standards.md § Error Envelope).

List Endpoints¶

These rules apply to every GET /api/v1/{resource} collection endpoint — paginated list, search, filter, sort. They were established in GRI-86 and are mandatory for every list endpoint we ship. The shared building blocks live in src/shared/api/pagination.py, src/shared/api/sorting.py, and src/shared/api/search.py.

The standard is shaped around three constraints we don't relitigate per resource:

1. The FE uses Orval to codegen useListItems(query) hooks + Zod schemas from our OpenAPI spec. Every filter must survive that codegen — i.e. it must be a typed scalar query param, not a custom DSL string. RSQL, filter[x]=, JSON-encoded filters are all out.
2. TanStack Table needs random-access pagination + a total count. That rules out cursor pagination as the default. Offset wins.
3. URL is the source of truth on the FE. Pagination, filters, sort, and snapshot are all in the URL search params. The "draft vs committed" state lives on the FE side (see grinsystem-front/docs/standards.md); the BE only sees the committed shape.

Two companion read patterns sit alongside the paginated list and follow further down in this file: § Stats endpoints (filter-agnostic, tenant-wide aggregate counts) and § List Resource Options Endpoints (bounded-cardinality picker / combobox shapes). All three patterns are mandatory where they apply — they're not interchangeable, and shoehorning one into another (e.g. populating a picker from the paginated list) is an anti-pattern called out in each section.

One Query() parameter per list endpoint (mandatory)¶

A list endpoint exposes exactly one Pydantic Query() parameter: a single model that inherits PaginationParams and carries pagination + filters + sort. Do not add any second Query() parameter alongside it — neither another Pydantic model nor a bare scalar (e.g. a standalone sort).

Why this is a hard rule. FastAPI (≥ 0.115) flattens a Pydantic Query() model into individual, per-field query params only when it is the sole query parameter on the handler. The moment a second Query() param appears — another model or a scalar — FastAPI stops flattening and emits each model as a nested $ref query param marked required: true. Consequences:

• The endpoint 422s on flat query strings (?page=1&status=active) and on no params at all — the wrapper object is reported "missing", so the per-field defaults never apply.
• The OpenAPI contract advertises required: true object params even though every field has a default, and the FE's Orval codegen emits unserializable nested params (?pagination=[object Object]).

The breakage is silent: code, types, and app startup all look fine — only a real request (or the FE) reveals it. Established in GRI-104.

# ❌ BROKEN — a second Query() param (another model OR a scalar) stops flattening
async def list_items(
    pagination: Annotated[PaginationParams, Query()],
    filters: Annotated[ItemListFilterParams, Query()],       # second model
    sort: Annotated[ItemSortToken | None, Query()] = None,   # ...or just a scalar
): ...

# ✅ CORRECT — one model carries pagination + filters + sort
async def list_items(
    params: Annotated[ItemListQueryParams, Query()],
): ...

Depends() is not an escape hatch. Annotated[Model, Depends()] flattens even with multiple models/scalars, but it silently drops the extra="forbid" unknown-param rejection (?bogus=1 → 200 instead of 422), so typos and stale FE params pass through unnoticed. Always use Query() with the single-model rule.

Pagination + snapshot¶

The single query model inherits PaginationParams from src.shared.api.pagination, so it gets page / page_size / as_of / .offset / .resolve_snapshot() for free:

from typing import Annotated
from fastapi import APIRouter, Query
from src.shared.api.pagination import PaginatedResponse, PaginationParams

class ItemListQueryParams(PaginationParams):
    # filters + sort fields — see § Filter parameters and § Sort
    ...

@router.get("/items")
async def list_items(
    params: Annotated[ItemListQueryParams, Query()],
) -> PaginatedResponse[ItemListItem]:
    snapshot = params.resolve_snapshot()
    items, total = await handler.handle(... , snapshot, params.offset, params.page_size)
    return PaginatedResponse.build(
        items=[ItemListItem.model_validate(x) for x in items],
        total=total,
        params=params,
        snapshot_at=snapshot,
    )

Defaults: page=1, page_size=20, max page_size=100. Boundary violations surface as Pydantic 422.

The as_of snapshot is optional on the wire. If absent, the route resolves now() once per request via params.resolve_snapshot(), passes it down to the repo (which applies WHERE created_at <= :as_of), and echoes it back in snapshot_at. If present, it's echoed back unchanged. The FE writes snapshot_at to the URL on first navigation (navigate({ replace: true })) so subsequent page changes reuse the same snapshot — consistent row set + stable total. The FE drops as_of on Apply Filters / Apply Sort / explicit Refresh so the BE re-snapshots.

Scope of the snapshot is intentional:

• as_of constrains created_at only. Updates (updated_at, is_active flips, name edits) flow through live — that's a feature; users want to see live status.
• Soft-deletes during browsing: a row visible on page 1 may be filtered out of a refresh of page 1 because the status filter excludes it; total stays correct because the snapshot still counts the row.
• For genuinely point-in-time reporting (regulatory, audit), build a dedicated history table — don't lean on as_of.

Filter parameters — fields on the single query model¶

Filters are fields on the one query model, not a separate model. The model inherits PaginationParams (so extra="forbid" and the pagination fields come for free) and declares the per-resource filters + sort:

from typing import Literal
from uuid import UUID
from datetime import datetime
from pydantic import Field
from src.shared.api.pagination import PaginationParams

class ItemListQueryParams(PaginationParams):
    # pagination (page / page_size / as_of) + extra="forbid" inherited

    q: str | None = Field(default=None, max_length=200)
    item_type: list[ItemType] = Field(default_factory=list)
    status: Literal["active", "inactive", "all"] = "active"   # see § Status filter
    source: Literal["tenant", "system", "all"] = "tenant"
    item_group_id: UUID | None = None
    created_after: datetime | None = None
    created_before: datetime | None = None
    sort: ItemSortToken | None = None                         # see § Sort

Repeated scalar params naturally become lists: ?item_type=ingredient&item_type=packaging.

extra="forbid" (inherited from PaginationParams) rejects unknown query params with 422 — typos and stale FE code don't get silently dropped. Do not split filters into a second Query() model or pull any field out into a standalone scalar param — that re-triggers the flattening bug (see § One Query() parameter per list endpoint).

Filter-agnostic, tenant-wide aggregate counts belong on the companion /stats endpoint, not on this filter-aware list response — see § Stats endpoints.

Status filter — tri-state enum, implicit default = working set¶

Every list endpoint that has an active/archived/closed distinction uses a tri-state enum, not a bool:

status: Literal["active", "inactive", "all"] = "active"     # items
status: Literal["open", "closed", "all"] = "open"           # orders
status: Literal["active", "archived", "all"] = "active"     # recipes

Why not active: bool | None? The unset case is ambiguous (omit → "all" or "default"?) and there's no clean way to express "all" without a literal null string. The tri-state makes all three cases explicit and gives a clean Zod enum on the FE.

Default is "working set" (active / open / etc.) — aligns with the domain's ubiquitous language ("show me the recipes" means "the recipes I work with", not "the full historical archive"). One-off admin pages may override the default if the dominant task differs (e.g., an audit log defaults to all).

BE/FE coordination: URL stays clean at default state — the FE only writes the param when the user picks a non-default value. Route.useSearch() is the FE source of truth for the form widgets, and its validateSearch zod schema must mirror the BE default so widgets render the right value when the URL is empty. URL and form display stay in sync via the schema, not by writing redundant defaults to the URL.

Sort — ?sort=field / ?sort=-field, single-sort, stable tiebreaker¶

JSON:API-style. The sort allowlist lives in two coordinated places:

1. A domain-layer StrEnum ({Noun}ListSortField) — the persistence-side allowlist used by the repo to look up the SQLAlchemy column.
2. An API-layer Literal[...] of "field" + "-field" tokens ({Noun}SortToken) — the wire-side allowlist that Pydantic uses to validate the sort= query param and to emit a typed enum into the OpenAPI schema (so the FE's Orval/Zod codegen gets autocomplete + validation).

Both lists share the same member names. sort is a field on the single query model (never a standalone scalar Query() param — that re-triggers the flattening bug, see § One Query() parameter per list endpoint). The route reads params.sort, splits the optional - prefix, and resolves the bare value against the domain enum:

from typing import Annotated, Literal
from enum import StrEnum
from fastapi import Query
from src.shared.api.pagination import PaginationParams
from src.shared.api.sorting import SortDirection, parse_sort

# domain-layer allowlist — lives next to the repo ABC
class ItemListSortField(StrEnum):
    NAME = "name"
    CODE = "code"
    CREATED_AT = "created_at"
    UPDATED_AT = "updated_at"
    ITEM_TYPE = "item_type"
    STATUS = "status"

# API-layer typed token — lives in api/schemas/{noun}_queries.py
ItemSortToken = Literal[
    "name", "-name",
    "code", "-code",
    "created_at", "-created_at",
    "updated_at", "-updated_at",
    "item_type", "-item_type",
    "status", "-status",
]

DEFAULT_SORT = (ItemListSortField.CREATED_AT, SortDirection.DESC)

# sort is a field on the one query model, alongside pagination + filters
class ItemListQueryParams(PaginationParams):
    # ... filters ...
    sort: ItemSortToken | None = Field(
        default=None, description="Sort field with optional '-' prefix."
    )

@router.get("/items")
async def list_items(
    params: Annotated[ItemListQueryParams, Query()],
):
    field, direction = parse_sort(params.sort, ItemListSortField, default=DEFAULT_SORT)
    ...

Pydantic enforces the allowlist at validation time → 422 before the route body runs. parse_sort still runs (it splits the - prefix and maps to the domain enum); its own InvalidSortFieldError branch is unreachable for FastAPI callers because Pydantic already validated, but the function stays defensive for internal callers and future shapes. Single-sort only; multi-sort (shift-click) is deferred until a real use case appears.

One enum per resource, no extra API/domain split. Don't define a second StrEnum at the API layer that mirrors the domain enum — the Literal[...] is the only API-side shape and the route imports the domain enum directly. A mapping table between two identical enums is dead code.

Stable tiebreaker is mandatory. The repo helper apply_sort always appends id <same-direction> to the ORDER BY:

from src.shared.api.search import apply_sort

stmt = apply_sort(
    select(ItemModel),
    sort_column=ItemModel.name,
    direction=direction,
    id_column=ItemModel.id,
)

Without it, the row order on ties is implementation-defined and shifts between requests — breaking offset pagination even before snapshot considerations enter. The tiebreaker is invisible at the API contract level; never bypass it.

TanStack Table column-header click is the only sort UI on the FE. No separate "Sort by" dropdown.

Search — single q param, case + diacritic-insensitive¶

One free-text param per resource, matching across a predefined field allowlist declared at the handler/repo level (not configurable from the wire). Use search_any from src.shared.api.search to combine unaccent_ilike across N columns into one OR clause — every list endpoint with a q parameter must use this helper, not an inline or_(...):

from src.shared.api.search import search_any

# Module-level constant — the per-resource search-column allowlist:
_Q_SEARCH_COLUMNS = (
    ItemModel.name,
    ItemModel.code,
    ItemModel.description,
    ItemGroupModel.name,
    func.array_to_string(ItemModel.tags, " "),   # array columns are flattened
)

# Inside the repo's filter-builder:
if filters.q:
    where_clauses.append(search_any(_Q_SEARCH_COLUMNS, filters.q))

Each scalar column in the allowlist gets a matching expression index in __table_args__ so the planner can use it:

Index("ix_item_name_search", func.lower(func.unaccent(name)))

The unaccent extension (and the IMMUTABLE promotion of its functions, required for expression indexes) is provisioned by the DB bootstrap init script deploy/postgres/init/01-create-roles.sh, not by a migration — CREATE EXTENSION / ALTER FUNCTION on a trusted extension's superuser-owned functions are outside the least-privilege migrations role's envelope. It's a one-time, repo-wide enablement, not a per-resource concern; migrations only create the lower(unaccent(col)) indexes.

search_any raises ValueError on an empty column list — callers must short-circuit on "no q" themselves before calling. Synthetic expressions (func.array_to_string(...), func.concat(...)) work as columns; they fall back to a seq scan rather than using an index, which is acceptable inside an already tenant-scoped query.

No FTS / tsvector yet — add when EXPLAIN shows ILIKE is the bottleneck.

Date-range filters — pair of scalar params¶

?created_after=2026-01-01&created_before=2026-05-01

Inclusive bounds. Either side optional (partial ranges work). Same pattern for any time-bounded filter (updated_*, deactivated_*, etc.). Two scalar datetime | None params keep the OpenAPI / Zod types clean — don't pack ranges into a single delimited string.

Page + count repo helper — fetch_page¶

Every list-repo runs the same two-step shape: a SELECT for the current page (with WHERE + ORDER BY + OFFSET/LIMIT) plus a parallel SELECT COUNT(*) against the same WHERE. Hand-rolling both queries per repo invites drift between them (the count predicate falls behind a new filter, the page query gets an extra clause the count misses, etc.). src.shared.api.listing.fetch_page centralises the pattern so the count-WHERE-must-match-page-WHERE invariant is enforced by construction:

from src.shared.api.listing import fetch_page
from src.shared.api.search import search_any

# Per-resource bits (filters + join shape) stay inside the repo:
_Q_SEARCH_COLUMNS = (ItemModel.name, ItemModel.code, ItemModel.description, ...)

def _build_list_filters(*, tenant_id, filters, as_of) -> list[ColumnElement[bool]]:
    where_clauses = [ItemModel.tenant_id == tenant_id, ItemModel.created_at <= as_of]
    if filters.q:
        where_clauses.append(search_any(_Q_SEARCH_COLUMNS, filters.q))
    # ... other per-resource filter blocks ...
    return where_clauses

async def list_items(self, *, ...) -> tuple[list[tuple[Item, str | None]], int]:
    where_clauses = _build_list_filters(...)
    join = ItemModel.__table__.join(ItemGroupModel, ..., isouter=True)
    rows, total = await fetch_page(
        self._session,
        stmt=select(ItemModel, ItemGroupModel.name).select_from(join),
        filters=where_clauses,
        sort_column=_SORT_COLUMNS[sort_field],
        id_column=ItemModel.id,
        descending=descending,
        offset=offset, limit=limit,
        count_from=join,         # COUNT runs against the join, not just the base table
    )
    return [(_to_domain(model), gn) for model, gn in rows], total

fetch_page appends the id tiebreaker to ORDER BY automatically (see § Sort), so the repo never has to remember it.

count_from is the FROM/JOIN expression the COUNT runs against:

• Plain single-table repo → count_from=Model (or Model.__table__).
• Repo that joins another table for filtering or projection → count_from=<the .join(...) expression> so the COUNT sees the same row set as the page. Don't pass the bare base table when a join is in play — the count will be wrong for any filter that references the joined columns.

fetch_page does NOT own per-resource concerns: which columns to filter on, which columns to project, which sort column corresponds to which enum member. Those stay in the repo. Only the orchestration shape — apply filters, sort+tiebreaker, page slice, parallel count — is shared.

Stats endpoints — mandatory companion to every list resource¶

Every paginated GET /api/v1/<resource> list endpoint MUST ship with a companion GET /api/v1/<resource>/stats endpoint returning tenant-wide, filter-agnostic aggregate counts. No exceptions — read-mostly, admin-only, and system-curated resources all qualify. The FE renders a StatsBar (total / active / archived / by-type counts) above the table that needs unfiltered numbers; without a dedicated endpoint the FE would have to fire N parallel list calls at page_size=1 per page render, which is wasteful and breaks under any concurrent-request cap.

For picker / combobox reads (a third, distinct shape — { id, label }[], no envelope, bounded cardinality), see § List Resource Options Endpoints. Don't fold picker shapes into /stats; the shapes, lifetimes, and invalidation surfaces are different.

The shape rules below are the contract — every stats endpoint matches it.

Shape contract

• One flat Pydantic response class per stats endpoint, model_config = ConfigDict(extra="forbid").
• Fields are scalars or one-level-nested breakdown sub-models (see below). No arrays of objects, no deeper nesting.
• A required total: int field on every stats response — every endpoint has it.
• Per-dimension breakdowns are named by_<dimension> (by_status, by_item_type, …) and typed as a dedicated sub-model whose fields are the enum values — NOT dict[str, int] and NOT dict[EnumType, int]. The response is exhaustive — every enum member is a concrete field, zero-count members included as 0, never null and never omitted. The wire JSON is identical to a plain map ({ "active": 118, "inactive": 6 }), but OpenAPI now emits concrete properties instead of a key-less additionalProperties, so FE codegen produces precise key types and faker-seeded mocks fill the real keys. This lets the FE render a static bar without conditional layout. A bare dict[str, int] erases the keys from the contract — see Forbidden anti-patterns.
• Generate the breakdown sub-model from the domain enum so the enum stays the single source of truth and the schema cannot drift — a new enum member adds its field automatically. Use the shared build_stats_breakdown(name, enum) helper from src.shared.api.stats (it wraps pydantic.create_model, pins extra="forbid", and absorbs the create_model typing friction in one place). (dict[EnumType, int] does NOT work: Pydantic v2 still serializes enum-keyed dicts as additionalProperties, so the keys never reach the contract.)
• No query parameters at all. No q, no item_type, no status — the endpoint always returns tenant-wide counts. Filter-aware aggregates belong in the paginated list response's metadata, not here (see § Filter parameters).
• No pagination, no sort, no cursor.

Example shape:

# src/modules/item_catalog/api/schemas/item_queries.py
from pydantic import BaseModel, ConfigDict

from src.modules.item_catalog.domain.value_objects import ItemStatus, ItemType
from src.shared.api.stats import build_stats_breakdown

# Breakdown sub-models generated from the domain enums → OpenAPI gets concrete
# properties (active/inactive, every ItemType), enum is the single source of truth.
ItemStatsByStatus = build_stats_breakdown("ItemStatsByStatus", ItemStatus)
ItemStatsByItemType = build_stats_breakdown("ItemStatsByItemType", ItemType)


class ItemStatsResponse(BaseModel):
    model_config = ConfigDict(extra="forbid")

    total: int
    # A generated model used as a field type trips mypy's `valid-type` check
    # (it's a value, not a statically-known class) — silence it on the field line.
    by_status: ItemStatsByStatus  # type: ignore[valid-type]
    by_item_type: ItemStatsByItemType  # type: ignore[valid-type]

The route constructs each sub-model from the handler's exhaustive enum-valued dict — by_status=ItemStatsByStatus(**result.by_status). The application/query layer keeps returning plain dict[str, int]; the enum→property mapping is purely an API/OpenAPI concern.

// GET /api/v1/items/stats →
{
  "total": 124,
  "by_status":    { "active": 118, "inactive": 6 },
  "by_item_type": { "ingredient": 80, "product": 22, "packaging": 14, "intermediate": 8, "consumable": 0, "service": 0, "resale": 0 }
}

Layering — standard query path: API → Handler → Repo → SQL. Mirrors the list-endpoint layering above; only the inputs and shape differ.

• Repo method named stats(tenant_id) returns a typed dataclass (e.g. ItemStats), never a raw dict. It issues exactly one SQL statement using GROUP BY and FILTER (WHERE …) clauses — never N+1 across dimension keys. Tenant isolation rides on RLS (see § Authorization), not a hand-rolled WHERE tenant_id.
• Handler lives in application/queries/get_<resource>_stats.py and exposes Get<Resource>StatsQuery + Get<Resource>StatsResult + Get<Resource>StatsHandler, following § CQS Pattern → Query Handler. Decorate the handler with @traced_handler.
• API route lives in the same api/routers/<resource>.py file as the list route. Declare GET /<resource>/stats BEFORE GET /<resource>/{id} — FastAPI matches routes in declaration order and stats would otherwise be swallowed as the {id} path param.

Authorization — gate the stats endpoint with the same Permission as the corresponding list endpoint (e.g. Permission.ITEM_READ for /items/stats). Do not introduce a separate <RESOURCE>_STATS_READ permission — anyone who can see the list can see its aggregate. See § Authorization for the broader permission-gating rules.

Caching — out of scope for v1. No cache layer, no Cache-Control header. A tenant-wide change feed + ETag is the eventual hook; revisit once a hot stats endpoint surfaces in profiling.

Required tests (per implementation, not for the convention itself):

• Repo test asserting exactly one SQL statement is issued for stats(tenant_id) — use the assert_num_queries-style helper or sqlalchemy.event.listen('after_cursor_execute') to count statements.
• Integration test seeding ≥1 row per enum value, asserting every dimension key appears with the correct count (including zero-count keys after deleting/skipping a value).
• RLS test: seeded rows in tenant A are invisible from tenant B (response is zeroed out).
• Schema-contract test asserting each breakdown sub-model's field set equals the enum's value set (set(ItemStatsByStatus.model_fields) == {m.value for m in ItemStatus}). With create_model this can't drift, but the test pins the contract so a future hand-rolled sub-model or a renamed enum member fails loudly. Lives in tests/unit/modules/<module>/api/test_stats_schemas.py.

Forbidden anti-patterns

• ❌ Typing a breakdown as dict[str, int] (or dict[EnumType, int]). Both render in OpenAPI as a key-less additionalProperties map, erasing the enum keys from the contract — FE codegen degrades to { [key: string]: number } and faker-seeded mocks emit random keys, so the StatsBar shows zeros (and historically crashed). Always a create_model-from-enum sub-model with concrete properties.
• ❌ Accepting any query parameter on the stats endpoint.
• ❌ Returning null for a dimension key — always 0, always exhaustive.
• ❌ Computing stats inside the list endpoint and folding them into the list response envelope. List total stays filter-aware (paginates the current view); stats counts stay filter-agnostic (a stable "what's in my catalog" indicator). Two separate concerns, two separate endpoints.
• ❌ N+1 — one SQL statement per dimension key. Always a single GROUP BY / FILTER (WHERE …) query.

Pattern, not framework — each entity declares its own /stats route with its own response shape that matches the banner. Don't try to build a generic stats engine ahead of demand (see "What we deferred" below).

List Resource Options Endpoints — picker shapes for combobox / select UIs¶

A third API pattern alongside § List Endpoints and § Stats endpoints. Picker / select / combobox UIs ask "what are the choices?" — and the answer is neither a paginated list nor a stats aggregate. Populating a picker from GET /api/v1/<resource> ships 10× more bytes than needed, forces an awkward page_size=1000-or-paginate decision, and couples the picker cache to the list-page cache (same query key → invalidation collisions even when the picker doesn't care about non-name changes).

When the endpoint is required

A resource MUST expose GET /api/v1/<resource>/options[?<scope>] when both of these are true:

• It is the target of a "pick one from a list" picker (combobox / select) on the FE.
• Per-tenant cardinality is bounded — practical cap < ~500 rows.

A resource MUST NOT expose /options (use the list endpoint with ?q= server-side typeahead instead) when:

• Cardinality is unbounded (items, customers, sales orders) — /options would balloon the payload.

A resource does not need /options at all when it never appears in a picker.

Shape contract

• One flat Pydantic class per options endpoint, model_config = ConfigDict(extra="forbid"), with exactly two fields:
• id: UUID
• label: str (the human-facing name)
• Bare JSON array response — list[<Resource>OptionResponse]. Not wrapped in an { items, total } envelope. The picker consumes the whole filtered list in one fetch; pagination is not a concept here.
• Optional scope params — may accept flat enum / UUID scope filters that mirror the resource's intrinsic partitioning (e.g. item_type for item-groups, warehouse_id for locations). Scope params follow the same flat-Pydantic-model rule as § Filter parameters.
• No q search, no sort, no page / page_size, no cursor. Picker filters the array locally via cmdk. Server-side typeahead resources don't qualify — they use the list endpoint.
• Sorted alphabetically by label server-side (stable, predictable for both UI scan and snapshot tests).
• Excludes archived / inactive rows by default — a picker shows pickable choices. Soft-deleted / deactivated entities never appear.

Example shape:

# src/modules/item_catalog/api/schemas/item_group_queries.py
from pydantic import BaseModel, ConfigDict


class ItemGroupOptionResponse(BaseModel):
    model_config = ConfigDict(extra="forbid")

    id: UUID
    label: str   # the group's name


class ListItemGroupOptionsQueryParams(BaseModel):
    model_config = ConfigDict(extra="forbid")

    item_type: ItemType | None = None   # optional scope filter

// GET /api/v1/item-groups/options?item_type=ingredient →
[
  { "id": "8d2d…-…-…", "label": "Drożdże" },
  { "id": "4a91…-…-…", "label": "Mąki" },
  { "id": "0c11…-…-…", "label": "Sole" },
  { "id": "1f70…-…-…", "label": "Słodziki" }
]

Layering — standard query path: API → Handler → Repo → SQL. Mirrors § Stats endpoints; only the inputs and shape differ.

• Repo method named list_options(tenant_id, *scope_filters) returns list[<Resource>Option] where <Resource>Option is a frozen dataclass with id + label. One SQL statement: SELECT id, name AS label FROM <table> WHERE … [scope] AND deleted_at IS NULL ORDER BY label. Tenant isolation rides on RLS (see § Authorization), not a hand-rolled WHERE tenant_id.
• Handler lives in application/queries/list_<resource>_options.py and exposes List<Resource>OptionsQuery + List<Resource>OptionsResult + List<Resource>OptionsHandler, following § CQS Pattern → Query Handler. Decorate the handler with @traced_handler.
• API route lives in the same api/routers/<resource>.py as the list route. Declare GET /<resource>/options BEFORE GET /<resource>/{id} — FastAPI matches routes in declaration order and options would otherwise be swallowed as the {id} path param. Same precedence pitfall as /stats.

Authorization — gate the options endpoint with the same Permission as the corresponding list endpoint (e.g. Permission.ITEM_VIEW for /item-groups/options). Do not introduce a separate <RESOURCE>_OPTIONS_READ permission — anyone who can browse the list can populate a picker. See § Authorization for the broader permission-gating rules.

Caching — out of scope for v1. No cache layer, no Cache-Control header. ETag is the eventual hook once a tenant-wide change feed is wired; revisit once a hot options endpoint surfaces in profiling.

Required tests (per implementation, not for the convention itself):

• Repo test asserting exactly one SQL statement is issued for list_options(tenant_id, …) — use the same assert_num_queries-style helper the /stats repo test uses.
• Integration test seeding rows across each scope value, asserting:
• only matching rows returned (scope filter works),
• alphabetical ordering by label,
• bare-array shape (no envelope),
• archived / inactive rows excluded (where the resource has soft-delete).
• RLS test: tenant A's options invisible from tenant B.

Forbidden anti-patterns

• ❌ Returning full domain objects (Group, Item, Customer, …) — keep the shape to { id, label }.
• ❌ Pagination, total, or any envelope wrapper.
• ❌ Server-side search (?q=) — push consumers onto the list endpoint for that.
• ❌ Routing /options through the same query handler as the list endpoint — different shapes, different cache lifetimes, different invalidation surface.
• ❌ Per-resource permission like ITEM_GROUP_OPTIONS_READ — reuse the list permission.

Query handler shape¶

List endpoints follow the existing query-handler pattern (see § CQS Pattern → Query Handler):

• application/queries/list_{noun}.py contains List{Noun}Query (dataclass), {Noun}ListReadModel (dataclass — flat row shape, never the domain aggregate), List{Noun}Handler decorated with @traced_handler.
• The API-layer Pydantic params (the single {Noun}ListQueryParams model — pagination + filters + sort, inheriting PaginationParams) map into the domain-layer List{Noun}Query in the route via params.to_filters() + the parsed sort tuple. Pydantic schemas don't leak into application/.
• The handler returns (list[{Noun}ListReadModel], total). The route builds the PaginatedResponse[T] envelope.
• The repo's list() method materialises read models directly — never hydrates the full aggregate. Reads bypass domain validation; writes go through the aggregate.

Permissions / RLS¶

• Every list endpoint gated with Depends(require_permission(Permission.X_VIEW)) — declare the permission + its PERMISSION_ROLES entry in the same commit as the endpoint.
• Tenant isolation via RLS — no manual WHERE tenant_id in the repo. The session-scoped current_setting('app.current_tenant') does the filtering.

What we deferred (and why)¶

PATCH Endpoint Conventions¶

These rules apply to every PATCH /api/v1/{resource}/{id} endpoint that exposes partial updates on a flat aggregate. They were established in GRI-13 (PATCH /items/{item_id}) and are mandatory for every future Edit* command.

When to use PATCH vs PUT vs POST-with-verb vs sub-resources¶

The parent aggregate's PATCH endpoint always edits scalar fields only. Child collections are managed via their own URLs.

The tri-state field problem¶

A PATCH endpoint must distinguish three client intents per editable field:

A naive field: T \| None = None collapses cases 1 and 3 — the server can't tell "leave EAN alone" from "delete the EAN". This breaks no-op detection, breaks audit trails, and breaks downstream consumers that key off old_value → new_value.

The shared UNSET sentinel (Pydantic-native)¶

src/shared/sentinels.py exports a singleton UNSET whose only purpose is to mean "this field was not provided." It is a third value distinct from None and from every user value, with its own type (_UnsetType) so mypy can narrow correctly.

The sentinel is Pydantic-aware: it implements __get_pydantic_core_schema__ and __get_pydantic_json_schema__ so Pydantic schemas can use T | _UnsetType = Field(default=UNSET, ...) directly. The route receives body.fieldname already set to either UNSET (field absent), None (explicit null), or the parsed value — no model_fields_set juggling needed.

# src/shared/sentinels.py — public API
UNSET: Final[_UnsetType]        # the singleton

# Pydantic integration on _UnsetType:
#   __get_pydantic_core_schema__   — UNSET cannot appear in user input;
#                                    only valid as a default when the field is absent
#   __get_pydantic_json_schema__   — returns {} so the sentinel does NOT pollute
#                                    the OpenAPI emission with a meaningless type

The OpenAPI emission point matters: client-side codegen and FE form generators consume the OpenAPI schema. The sentinel must be invisible there — clients should see the field as "optional with a normal type", not "optional with a mystery _UnsetType arm in a union".

PATCH schema pattern¶

The Pydantic schema is the single source of truth for what the endpoint accepts. It only declares mutable fields; immutable fields are rejected by a class-level validator with the right 400 code, so they never appear in the OpenAPI emission and never show up in FE-generated forms.

Every PATCH schema inherits from PatchRequestModel (in src/shared/api/schemas.py), which centralises the cross-cutting rules so every endpoint behaves identically and no schema can accidentally skip them.

The shared base — PatchRequestModel¶

# src/shared/api/schemas.py
from typing import Any, ClassVar

from pydantic import BaseModel, ConfigDict, model_validator

from src.shared.exceptions import ImmutableFieldCannotBeChangedError


class PatchRequestModel(BaseModel):
    """
    Base class for every ``PATCH`` endpoint's request schema.

    Centralises three rules that apply to every PATCH endpoint
    (see ``docs/architecture/conventions.md`` § PATCH Endpoint Conventions):

    1. ``extra="forbid"`` — unknown fields produce 422 ``VALIDATION_ERROR``.
    2. Mutable fields are declared explicitly by the subclass with
       ``T | _UnsetType = Field(default=UNSET, ...)`` (or
       ``T | None | _UnsetType`` for clearable fields).
    3. Immutable fields are listed in ``_IMMUTABLE_FIELDS`` (a class-level
       ``frozenset[str]``). The ``model_validator`` below rejects any
       attempt to send them with **400** ``IMMUTABLE_FIELD_CANNOT_BE_CHANGED``.

    Subclasses with no immutable fields simply omit the declaration — the
    empty default ``frozenset()`` means "nothing is off-limits."
    """

    model_config = ConfigDict(extra="forbid")

    _IMMUTABLE_FIELDS: ClassVar[frozenset[str]] = frozenset()

    @model_validator(mode="before")
    @classmethod
    def _reject_immutable_fields(cls, data: Any) -> Any:
        """Raise the domain 400 if the client attempted to mutate any immutable field."""
        if isinstance(data, dict):
            attempted = cls._IMMUTABLE_FIELDS & set(data.keys())
            if attempted:
                raise ImmutableFieldCannotBeChangedError(
                    f"Immutable fields cannot be changed: {sorted(attempted)}.",
                )
        return data

The base lives in src/shared/api/schemas.py (next to the other shared API utilities like deps.py and router.py). The ImmutableFieldCannotBeChangedError lives in src/shared/exceptions.py since the concept is use-case-agnostic — every module's Edit* command can raise it.

A concrete subclass¶

# src/modules/item_catalog/api/schemas/items.py
from typing import ClassVar

from pydantic import Field, field_validator

from src.modules.item_catalog.domain.value_objects.tags import MAX_TAG_COUNT, MAX_TAG_LENGTH
from src.shared.api.schemas import PatchRequestModel
from src.shared.sentinels import UNSET, _UnsetType


class EditItemRequest(PatchRequestModel):
    """Request body for ``PATCH /api/v1/items/{item_id}``."""

    _IMMUTABLE_FIELDS: ClassVar[frozenset[str]] = frozenset({"code", "item_type", "storage_unit"})

    # Mutable, non-nullable fields — sentinel-or-value:
    name: str | _UnsetType = Field(default=UNSET, min_length=1, max_length=200)
    tags: list[str] | _UnsetType = Field(default=UNSET, max_length=MAX_TAG_COUNT)

    # Mutable, nullable (clearable) fields — sentinel-or-value-or-null:
    description: str | None | _UnsetType = Field(default=UNSET, max_length=500)
    shelf_life_days: int | None | _UnsetType = Field(default=UNSET, gt=0)
    ean: str | None | _UnsetType = Field(default=UNSET, min_length=1, max_length=20)

    @field_validator("tags")
    @classmethod
    def _enforce_per_tag_length(cls, value: list[str] | _UnsetType) -> list[str] | _UnsetType:
        """Reject any tag longer than ``MAX_TAG_LENGTH`` characters (skip if UNSET)."""
        if isinstance(value, _UnsetType):
            return value
        for tag in value:
            if len(tag) > MAX_TAG_LENGTH:
                raise ValueError(f"Each tag must be at most {MAX_TAG_LENGTH} characters.")
        return value

The subclass shape is now: _IMMUTABLE_FIELDS declaration + field declarations + any per-field validators. No model_config, no model_validator — both inherited.

Why the immutable check is in the schema, not the handler:

1. The schema is what produces the OpenAPI spec. Front-end form generators consume it. If the schema declares code, FE forms will show a code field that the BE will then reject — confusing UX.
2. The rejection still happens before any I/O (it's model_validator(mode="before") on the base).
3. The shared exception handler maps ImmutableFieldCannotBeChangedError to the contract-mandated 400 with IMMUTABLE_FIELD_CANNOT_BE_CHANGED. Pydantic's own "extra inputs are not permitted" would produce 422 with the generic VALIDATION_ERROR — wrong status, wrong code, wrong UX.
4. extra="forbid" is still on for genuinely unknown fields (typos, junk) — those continue to produce 422 as expected.

Nullable vs non-nullable decision per field:

• Non-nullable (T | _UnsetType): the field can never be cleared on this aggregate. Sending null is a client bug. Pydantic will raise a type error because the schema's union doesn't include None. The client sees 422 with a useful Pydantic message.
• Nullable (T | None | _UnsetType): the field can be cleared. Sending null is the valid way to set it back to NULL in the DB.

Checklist when writing a new Edit* schema:

1. Inherit from PatchRequestModel (NOT BaseModel).
2. Declare _IMMUTABLE_FIELDS: ClassVar[frozenset[str]] = frozenset({...}) listing every field that cannot change after creation. Omit the declaration if nothing is immutable.
3. Declare each mutable field with T | _UnsetType (non-nullable) or T | None | _UnsetType (clearable), defaulting to UNSET via Field(default=UNSET, ...).
4. Add per-field validators only for constraints that go beyond the standard min_length / max_length / gt / etc — e.g. per-element checks on a list.
5. Do not redeclare model_config = ConfigDict(extra="forbid") — it's inherited.
6. Do not add your own _reject_immutable_fields validator — it's inherited.

Command + handler + domain layers¶

Because the schema does the field-absence translation, every layer below it sees the same T | _UnsetType (or T | None | _UnsetType) and never needs to know about the HTTP-side specifics.

# application/commands/edit_item.py
@dataclass
class EditItemCommand:
    """Inputs for UC-002 EditItem. No immutable-field plumbing — the schema handles that."""

    item_id: UUID
    tenant_id: UUID
    updated_by: UUID
    name: str | _UnsetType = UNSET
    description: str | None | _UnsetType = UNSET
    tags: list[str] | _UnsetType = UNSET
    shelf_life_days: int | None | _UnsetType = UNSET
    ean: str | None | _UnsetType = UNSET


@dataclass
class EditItemResult:
    item_id: UUID
    changed: bool
    version: int


class EditItemCommandHandler:
    async def handle(self, command: EditItemCommand) -> EditItemResult:
        item = await self._item_repo.get_by_id(command.item_id, command.tenant_id)
        if item is None:
            raise ItemNotFoundError(...)
        if item.is_system:
            raise SystemItemIsReadOnlyError(...)

        # EAN uniqueness pre-check (only when the client actually sent a non-null EAN):
        if not isinstance(command.ean, _UnsetType) and command.ean is not None:
            existing = await self._item_repo.get_by_ean(command.ean, command.tenant_id)
            if existing is not None and existing.id != item.id:
                raise EANAlreadyInUseError(...)

        old_ean = item.ean.value if item.ean is not None else None
        changed = item.edit(
            updated_by=command.updated_by,
            name=command.name,
            description=command.description,
            tags=command.tags,
            shelf_life_days=command.shelf_life_days,
            ean=command.ean,
        )
        if not changed:
            return EditItemResult(item_id=item.id, changed=False, version=item.version)

        await self._item_repo.save(item)
        await self._event_publisher.publish(<entity>Updated(...))
        return EditItemResult(item_id=item.id, changed=True, version=item.version)

Canonical comparison idiom in the domain method:

def edit(self, *, name: str | _UnsetType = UNSET, ..., updated_by: UUID) -> bool:
    changed = False
    if not isinstance(name, _UnsetType) and name != self.name:
        self.name = name
        changed = True
    # … one block per editable field …
    if changed:
        self.updated_at = datetime.now(tz=timezone.utc)
        self.updated_by = updated_by
    return changed

The route is a pure pass-through¶

Because the schema does the tri-state translation and rejects immutable fields, the route has nothing to do but map fields onto the command:

# api/routers/items.py
@items_router.patch("/{item_id}", status_code=200, response_model=EditItemResponse)
@idempotent
async def edit_item(
    item_id: UUID,
    body: EditItemRequest,
    handler: EditItemHandlerDep,
    user: Annotated[CurrentUser, Depends(require_permission(Permission.ITEM_UPDATE))],
) -> EditItemResponse:
    """UC-002: partial-update an Item using the UNSET sentinel pattern."""
    result = await handler.handle(
        EditItemCommand(
            item_id=item_id,
            tenant_id=user.tenant_id,
            updated_by=user.user_id,
            name=body.name,
            description=body.description,
            tags=body.tags,
            shelf_life_days=body.shelf_life_days,
            ean=body.ean,
        )
    )
    return EditItemResponse(item_id=result.item_id, changed=result.changed, version=result.version)

No model_fields_set, no assertions, no immutable-set extraction. The route honours CLAUDE.md's "routes are thin" rule literally — it only maps fields.

OpenAPI / FE form generation¶

The schema's emitted OpenAPI schema is what front-end form generators consume. Because the schema only declares mutable fields, the generated FE form only contains mutable fields — no code/item_type/storage_unit boxes for the user to fight with.

If you ever need a field to appear in OpenAPI but be rejected, do not add it to the schema. Instead either (a) extend _IMMUTABLE_FIELDS and document the reason, or (b) extend _IMMUTABLE_FIELDS and add a description= to the OpenAPI docs explaining why the field is read-only on this verb. The current default — don't show what can't be sent — is preferred unless there's a concrete reason.

No-op detection contract¶

A pure no-op PATCH (no field changed) must:

• Return 200 with {"changed": false}.
• Skip repo.save() — no UPDATE, no version bump, no updated_at/updated_by change.
• Skip event_publisher.publish() — no outbox row, no event dispatched.

The domain method is the single source of truth for "did anything change." Its signature returns bool:

def edit(self, *, …) -> bool: ...    # True = at least one field changed

A no-op edit and an absent field are semantically equivalent: submitting {"name": "<current value>"} produces the same outcome as submitting {}. Both return changed: false.

version in the PATCH response¶

Every PATCH response body includes the post-update version:

{ "item_id": "…", "changed": true, "version": 5 }

Returning the new version lets the FE chain edits without an extra GET. On a no-op, return the current version unchanged. This pairs with OptimisticLockMixin — the FE keeps the latest version and the BE rejects stale versions with 409 CONCURRENT_MODIFICATION (already wired in SqlAlchemyItemRepository.save()).

A future iteration may add ETag / If-Match headers as a stricter alternative to embedding version in the body — don't add it speculatively.

Required tests for every PATCH endpoint¶

Beyond the standard create-endpoint coverage, every PATCH endpoint must include:

• No-op test (integration) — submit {} (or values matching current) and assert 200 { "changed": false } plus zero new outbox rows for the aggregate.
• Sentinel-clears test (integration) — submit explicit null for a nullable field and assert the field is cleared in the DB, changed: true, exactly one outbox row.
• Immutable-field test (integration) — submit each immutable field and assert 400 IMMUTABLE_FIELD_CANNOT_BE_CHANGED. Parametrize over _IMMUTABLE_FIELDS to cover the whole set in one test. This lives at the integration layer because the rejection happens in the Pydantic schema, not in the handler — there's nothing to unit-test below it.
• Unknown-field test (integration) — submit a junk field and assert 422 VALIDATION_ERROR (proves extra="forbid" is still active).
• Optimistic concurrency test (integration) — stale version produces 409 CONCURRENT_MODIFICATION.

The handler-level unit tests still cover: happy path with field changes, no-op detection, system-item rejection, EAN-uniqueness conflict, not-found, domain invariants like shelf-life applicability. They do not test the immutable-field rejection (no longer reachable at this layer).

Database Conventions¶

Table Naming¶

• snake_case for tables and columns
• singular for table names (recipe, not recipes)
• id for primary key

Column Naming¶

• Foreign keys: {table}_id (e.g., recipe_id)
• Timestamps: {action}_at (e.g., created_at, published_at)
• Boolean: is_{state} (e.g., is_active)

Schema¶

CREATE TABLE recipe (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    code VARCHAR(50) NOT NULL,
    name VARCHAR(255) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'draft',
    current_version INT NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ,
    created_by UUID NOT NULL,
    updated_by UUID,

    CONSTRAINT uq_recipe_code_tenant UNIQUE (tenant_id, code)
);

-- RLS — baseline for new tables
ALTER TABLE recipe ENABLE ROW LEVEL SECURITY;
ALTER TABLE recipe FORCE ROW LEVEL SECURITY;            -- applies to the table owner too
CREATE POLICY tenant_isolation ON recipe
    USING      (tenant_id = current_setting('app.current_tenant')::uuid)
    WITH CHECK (tenant_id = current_setting('app.current_tenant')::uuid);

RLS pattern notes. Plain ENABLE is bypassed by the table owner — and the dev DB role (grinsystem) is currently a Postgres superuser with BYPASSRLS, so policies defined with ENABLE alone never actually filter in dev or tests. FORCE closes the owner bypass; running as a non-superuser, non-BYPASSRLS role (separately tracked) closes the superuser bypass. WITH CHECK mirrors USING for INSERT/UPDATE so writes can't slip a row with a foreign tenant_id past the policy.

Every table ships this tightened baseline. The migration chain was squashed pre-release into three revisions — 0001_baseline (empty anchor), 0002_platform_foundation (shared outbox), 0003_item_catalog_foundation (item + item_group) — all using ENABLE + FORCE + USING + WITH CHECK.

SQLAlchemy Mixins¶

The shared columns above are composed into every aggregate-root model via mixins in src/shared/db/mixins.py. Do not redeclare these columns per model — that's how naming, nullability, and timezone settings drift.

from src.shared.db.base import Base
from src.shared.db.mixins import (
    AuditedByMixin, OptimisticLockMixin, TenantScopedMixin, TimestampMixin, UUIDPkMixin,
)

class RecipeModel(
    UUIDPkMixin,
    TenantScopedMixin,
    TimestampMixin,
    AuditedByMixin,
    OptimisticLockMixin,
    Base,
):
    __tablename__ = "recipe"
    code: Mapped[str] = mapped_column(String(50), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    # ...domain-specific columns only

Domain entities do NOT inherit these mixins — domain stays pure-Python. The fields are mirrored in the dataclass; the convention is documentation, not enforcement.

Optimistic Concurrency Control¶

OptimisticLockMixin is the default for every mutable aggregate, not the exception. Add it unless you can name a specific reason not to (purely immutable rows like Invoice once issued, or strictly single-writer rows like an in-flight import staging table).

Why default-on: two people editing the same record at the same time is a normal occurrence in a multi-user ERP. Without versioning the second save silently overwrites the first. With versioning the second save raises ConcurrentModificationError and the API returns 409, the client refreshes, the user re-applies their edit.

End-to-end flow when this is enabled — must be implemented consistently across entity / model / repo / handler / API:

1. Model — composes OptimisticLockMixin. A freshly persisted row is version = 0; SQLAlchemy bumps version by 1 on every UPDATE and raises StaleDataError if zero rows match (id=X AND version=old). The 0 start is guaranteed by the mixin's custom version_id_generator — SQLAlchemy's built-in generator would otherwise stamp 1 on the first INSERT, contradicting both the entity's version=0 default and the column's server_default="0".
2. Entity — mirrors a version: int field. The repo populates it on get_by_id; the API returns it in every read; the client sends it back on every write.
3. Repository — translates StaleDataError from flush() into ConcurrentModificationError (HTTP 409). Writes the post-flush model.version back onto the entity so the handler/API can return the new version to the client.
4. API — exposes version on every read response (e.g. ItemRead.version) and on every write response so the client always has the post-write version to send next time. Every editable-aggregate write request body declares version: int as a required field (no default); a missing or non-integer version returns 422 at the Pydantic boundary. Body-less endpoints (POST /{id}/deactivate, POST /{id}/reactivate, DELETE /{id}/group) accept a minimal { "version": int } JSON body — the request body is the single, uniform OCC transport across the whole API (no If-Match header).
5. Handler — after loading the aggregate (and after the not-found / actor-aware authorization guards), compares command.version against the loaded entity.version. A mismatch raises ConcurrentModificationError before any expensive cross-module port call, before any state-machine pre-check, and before any domain mutation. The version check is a request-scoped concurrency guard (not a state-machine invariant), so it lives in the handler — the entity has no check_version() method. Use the same wording as the repo's StaleDataError branch so the two error messages stay aligned: "{Aggregate} '{id}' was modified by another writer; refresh and retry."
6. Error — ConcurrentModificationError(ConflictError) in src/shared/exceptions.py. The shared exception handler renders it as 409 { "error": { "code": "CONCURRENT_MODIFICATION", ... } }. Client retry strategy: refresh the resource, re-apply the edit, re-submit with the new version.

Precedence — when multiple guards could fire on the same request:

404 (not-found) → 403 (actor-aware, e.g. is_system) → 409 (stale version) → 400 (state-machine pre-check, where the precedence-exception in § Guard Placement applies) → cross-module guards → domain mutation.

A stale view makes every later check meaningless, so 409 stale-version wins over 400 state-machine and over any later 409 cross-module conflicts on the same request. 404 / 403 are more fundamental rejections (the resource doesn't exist or can't be edited at all) so they precede the version check. Each handler with multiple failure modes gets one precedence test asserting this ordering (configure two failures at once, assert stale-version wins).

Aggregates that do use it (Phase 1): - ItemCatalog: Item, ItemGroup (multiple editors per tenant) - Manufacturing: Recipe, ProductionOrder (collaborative editing + status transitions) - Warehouse: LOT, StockEntry (reservations vs consumption races) - CRM: Customer, SalesOrder (sales rep + customer service)

Aggregates that do not need it: - Identity: AuditLogEntry (append-only) - Manufacturing: RecipeNutritionSnapshot, ProductionCostRecord (immutable once written) - Invoicing: Invoice (immutable once issued), CreditNote - Traceability: BatchGenealogyLink, LotMovementRecord, Recall (append-only)

Indexes¶

Declare indexes on the SQLAlchemy model, not in the migration. That way make migrate-new autogenerate sees them and emits the right op.create_index(...) automatically — same source of truth, no drift.

from sqlalchemy import Index, func

class ItemModel(...):
    __tablename__ = "item"
    __table_args__ = (
        # plain composite indexes
        Index("ix_item_tenant_id_item_type", "tenant_id", "item_type"),
        Index("ix_item_tenant_id_item_group_id", "tenant_id", "item_group_id"),
        # expression index — autogenerate emits this correctly in Alembic >= 1.11
        Index("uq_item_tenant_id_code", "tenant_id", func.lower(text("code")), unique=True),
    )

Single-column indexes can also be inlined with mapped_column(..., index=True) — fine for tenant_id from TenantScopedMixin. Composite and expression indexes always go in __table_args__.

What you still hand-edit in the migration: - ENABLE + FORCE ROW LEVEL SECURITY and CREATE POLICY tenant_isolation with USING and WITH CHECK (see § Schema) - CHECK constraints with non-trivial expressions - Data backfills

Everything else — including expression-based unique indexes — belongs on the model.

Migrations¶

Always autogenerate. Never hand-write a migration from scratch.

make migrate-new M="add recipe nutrition snapshot"  # runs alembic revision --autogenerate

Hand-edit the generated file only for: - ENABLE + FORCE ROW LEVEL SECURITY and CREATE POLICY tenant_isolation with USING and WITH CHECK (see § Schema) - CHECK constraints with non-trivial expressions - Data backfills

Why: autogenerate uses Base.metadata + NAMING_CONVENTION directly, so constraint and index names (pk_*, fk_*, ix_*, uq_*) stay in sync with the model. Hand-written migrations drift. Indexes (including expression indexes) live on the model — see § Indexes below.

Seeding¶

Sample/dev data is never put in migrations — schema (Alembic) and data (seeders) stay separate so migrations remain deterministic and alembic check-able. Seeders are env-gated, idempotent, per-module, and run as a one-shot (python -m src.shared.seeding). See docs/architecture/seeding.md for the full standard, including how to add a seeder for a new module.

UUID Generation¶

Every aggregate ID uses UUIDv7 — time-ordered. Generate via the shared helper:

from src.shared.ids import new_id

item = Item(id=new_id(), ...)

Why not v4: v4 is random, so consecutive inserts hit random B-tree pages, fragmenting indexes and slowing inserts under load. v7's leading timestamp makes inserts monotonically increasing → tight indexes, better cache behaviour, ~10–20% throughput on write-heavy tables. The change is invisible to consumers (still 128-bit UUIDs over the wire). Never call uuid4() directly — always go through new_id().

Event Conventions¶

Topic Naming¶

grinsystem.{module}.events

Event Naming¶

• Past tense: RecipeCreated, OrderCompleted
• PascalCase

Event Structure¶

{
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "event_type": "RecipeCreated",
  "aggregate_id": "123e4567-e89b-12d3-a456-426614174000",
  "aggregate_type": "Recipe",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "timestamp": "2026-03-30T10:00:00Z",
  "version": 1,
  "payload": {
    "name": "Bread",
    "code": "BRD-001"
  },
  "metadata": {
    "correlation_id": "22222222-2222-2222-2222-222222222222",
    "causation_id": "33333333-3333-3333-3333-333333333333"
  }
}

Transactional Outbox (Reliable Event Publishing)¶

Every module that publishes events writes them through a transactional outbox, never directly. This is non-negotiable — at-least-once delivery is a foundational guarantee of the modular monolith, and dispatching an event straight from a command handler would lose it whenever the post-commit dispatch failed.

There is no message broker. Events are delivered in-process: the outbox row is the durable hand-off, the OutboxPoller is the relay, and the EventDispatcher invokes the @on_event handlers. Kafka is deferred until a module is extracted to its own service (see § Forward migration below).

How it works¶

1. The command handler writes the aggregate change and an outbox row in the same SQL transaction via OutboxEventPublisher. Either both land or neither does. The publisher also issues NOTIFY outbox_new so the poller wakes promptly.
2. The OutboxPoller (src/shared/outbox/poller.py) claims a batch of pending rows with SELECT ... FOR UPDATE SKIP LOCKED (so any number of pollers can run safely), reconstructs each DomainEvent from the row, and hands it to the EventDispatcher.
3. The EventDispatcher (src/shared/events/dispatcher.py) invokes every handler registered for type(event) — each event dispatched in its own fresh AsyncSession, so one handler's rollback can't poison the batch's marking transaction.
4. The poller marks each row completed (all handlers returned) or failed (a handler raised — error + retry_count recorded), then commits. Completed rows are pruned later (see § Retention).

Between batches the poller waits on Postgres LISTEN outbox_new for low-latency wake-up, falling back to a configurable timeout so a missed NOTIFY still drains. Lifespan wiring (load_all_handlers() + the poller task) lives in main.py.

Shared outbox table¶

One table, shared by every module. Defined in src/shared/outbox/models.py::OutboxEvent (composes UUIDPkMixin, TenantScopedMixin, TimestampMixin — append-only, so no OptimisticLockMixin / AuditedByMixin).

Columns:

id              UUID PRIMARY KEY                              -- from UUIDPkMixin
tenant_id       UUID NOT NULL                                  -- from TenantScopedMixin (+ index)
aggregate_type  VARCHAR(50) NOT NULL                           -- e.g. "Item", "Recipe"
aggregate_id    UUID NOT NULL
event_type      VARCHAR(80) NOT NULL                           -- e.g. "ItemCreated"; selects the handler set
payload         JSONB NOT NULL                                 -- the event's `payload` field ONLY (model_dump, mode="json"), not the full envelope — envelope fields live in their own columns above
metadata        JSONB NOT NULL                                 -- correlation_id, causation_id, headers (Python attr: event_metadata)
status          outbox_status NOT NULL DEFAULT 'pending'       -- pending → completed | failed
retry_count     INTEGER NOT NULL DEFAULT 0                      -- bumped each time a row is marked failed
error           VARCHAR(1000) NULL                             -- truncated str(exc) of the failing handler
processed_at    TIMESTAMPTZ NULL                               -- set when the row reaches completed/failed
trace_context   JSONB NULL                                     -- W3C traceparent/tracestate captured at publish; parents the consumer span
created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()             -- from TimestampMixin
updated_at      TIMESTAMPTZ NULL                               -- from TimestampMixin; stays NULL on this append-only table

Indexes (declared in __table_args__ on the model — autogenerate picks them up):

• ix_outbox_tenant_id (single column, from TenantScopedMixin)
• ix_outbox_tenant_id_created_at — supports per-tenant scans
• ix_outbox_unprocessed — partial index on (created_at) WHERE status = 'pending'; this is the index the poller's claim query rides, kept narrow to only the dispatch candidates.

Row level security is ENABLE + FORCE (so the table owner is also subject to the policy) and the tenant_isolation policy carries both USING and WITH CHECK clauses against current_setting('app.current_tenant')::uuid — see § Schema above for why this is the safer baseline for new tables. The poller drains all tenants' rows via the grinsystem_jobs (BYPASSRLS) role.

Application-side contract¶

Command handlers publish through OutboxEventPublisher (in src/shared/outbox/publisher.py), which only writes to the outbox table within the active session. The interface matches the EventPublisher protocol (async def publish(event: DomainEvent) -> None) so handlers stay transport-agnostic — the session is the transport.

Wiring: src/shared/api/deps.py exposes EventPublisherDep, an Annotated[EventPublisher, Depends(get_event_publisher)] alias that builds OutboxEventPublisher(session) per request. Command handlers depend on EventPublisherDep, never on a concrete class.

# command handler signature
async def handle(
    cmd: CreateItemCommand,
    session: DbSession,
    publisher: EventPublisherDep,
) -> ItemRead: ...

Retention¶

Completed outbox rows are deleted by the prune CLI (python -m src.shared.outbox.prune); failed rows are preserved for investigation. Default retention is 90 days post-processed_at. The CLI connects via JOBS_DB_URL (the grinsystem_jobs role, which has BYPASSRLS) so the cross-tenant DELETE actually sees every row. See outbox-retention.md for the full operational runbook. For the dispatch loop itself, see outbox-dispatcher.md.

Forward migration¶

When a module is extracted to its own service, the in-process dispatcher target is swapped for a Kafka producer on the publish side and a Kafka consumer on the subscribe side; per-module topics (grinsystem.<context>.events) reappear at that point. The outbox pattern and the @on_event handler signatures stay identical — only the transport between "committed outbox row" and "handler invocation" changes.

Event Handlers (@on_event)¶

Cross-module reactions are written as event handlers, never as direct calls into another module. A handler subscribes to one concrete DomainEvent subclass and runs when the dispatcher delivers that event.

Where handlers live + autodiscovery¶

Put a module's handlers in src/modules/<module>/application/event_handlers.py. At startup main.py calls load_all_handlers() (src/shared/events/autoload.py), which walks src/modules/*/application/event_handlers.py and imports each one, triggering the decorator registrations. Modules without that file are silently skipped; a transitive ImportError inside an existing handler module re-raises so a typo is never swallowed.

The decorator¶

@on_event(EventType) (src/shared/events/decorator.py) registers the decorated coroutine into the default dispatcher for that exact event class. It returns the handler unchanged, so a test can await my_handler(event, session) directly.

# src/modules/warehouse/application/event_handlers.py
from sqlalchemy.ext.asyncio import AsyncSession
from src.shared.events import on_event
from src.modules.item_catalog.domain.events import ItemCreated

@on_event(ItemCreated)
async def project_item_into_warehouse(event: ItemCreated, session: AsyncSession) -> None:
    # apply the side effect using the session the dispatcher passes in
    ...

Registration is keyed by exact class identity (type(event)): a handler on the broad DomainEvent envelope does NOT act as a wildcard. An event with no registered handler is marked completed with no side effects.

Per-handler session isolation¶

The dispatcher passes each handler a fresh AsyncSession minted by the poller per event — handlers never share the poller's lock-holding session. The handler does its work in that session; the poller commits it on success and rolls it back on failure. Do not open your own session inside a handler, and do not commit/rollback the passed session yourself.

When an event has multiple handlers they run concurrently under asyncio.gather with return_exceptions=False: the first failure propagates (and peers are cancelled), the poller marks the row failed, and the row is retried on a later batch. Write handlers so concurrent peers don't depend on each other's writes.

Idempotency expectation¶

Dispatch is at-least-once (a failed row is re-dispatched), so every handler must be idempotent — dedupe against a processed_events row in the same transaction as its side effects. See § Idempotency Keys → Handler-side idempotency.

Idempotency Keys¶

Two completely separate concerns, both mandatory:

A. Write-API idempotency (client → server)¶

Every state-mutating endpoint (POST / PUT / PATCH / DELETE) must accept the Idempotency-Key HTTP header (RFC draft, GitHub / Stripe / AWS standard). Clients supply an opaque token (UUIDv4 or anything ≤ 255 chars) per logical operation; the server caches (tenant_id, key) → (fingerprint, status, body) in Redis for 24 hours. A retry with the same key and the same request payload returns the cached response without re-executing the handler.

How to use — single decorator per endpoint:

from src.shared.idempotency import idempotent

@router.post("/items", status_code=201, response_model=ItemRead)
@idempotent
async def create_item(
    cmd: CreateItemCommand,
    handler: CreateItemHandlerDep,
) -> ItemRead:
    return await handler.handle(cmd)

Order matters: @router.post(...) must sit above @idempotent so FastAPI registers the wrapped function. The decorator handles header validation, fingerprinting, Redis lookup, cached-response replay, conflict detection, and the post-handler cache write — endpoints do nothing else.

Where to apply it:

Error mapping:

Fingerprinting: sha256(METHOD + "\n" + PATH + "\n" + raw_body_bytes). Method and path are baked in, so the same key sent to a different endpoint 409s cleanly. Sized responses > 64 KB skip the cache (logged at WARNING) — the handler runs idempotently on the next retry.

Storage: Redis. Key: idem:{tenant_id}:{key}. Value: {"fingerprint": "<hex>", "status": <int|null>, "body": <jsonable|null>}. An in-flight placeholder (status null) is set with a 30-second TTL to serialise concurrent retries; on handler success it is overwritten with the final entry and a 24h TTL. On handler exception the placeholder is deleted so the client can retry cleanly.

Trade-off: the Redis write is not atomic with the DB write — a crash between DB commit and the Redis update lets a retry re-execute the handler. This is the industry-standard compromise (Stripe accepts it). If you need strict atomic dedupe for a specific endpoint, layer a domain-level uniqueness constraint (e.g. a client_request_id column on the aggregate) on top.

Where it sits in code: src/shared/idempotency/decorator.py (the decorator) and src/shared/idempotency/redis.py (the async client + lifespan disposal). Command handlers know nothing about it.

B. Handler-side idempotency¶

The outbox dispatch loop is at-least-once: a row whose handler raises is marked failed and re-dispatched on a later batch. So every @on_event handler must dedupe before applying side effects.

Storage: each handler dedupes against a processed_events table (in the consuming module's schema). One row per event the handler has applied:

event_id       UUID PRIMARY KEY        -- from DomainEvent.event_id
consumer_name  VARCHAR(80) NOT NULL    -- which handler processed it
processed_at   TIMESTAMPTZ NOT NULL DEFAULT NOW()

Pattern (inside the session the dispatcher passes in):

@on_event(ItemCreated)
async def project_item(event: ItemCreated, session: AsyncSession) -> None:
    if await already_processed(event.event_id, session):
        return
    # ... apply side effects ...
    await mark_processed(event.event_id, session)

The dedupe check + mark + side-effect all run in one transaction (the dispatcher commits the handler's session on success). A crash after side-effect-commit-before-mark is impossible because the mark is in the same transaction.

Not yet built. No module wires @on_event handlers today, so there is no shared dedupe helper in src/shared/events/ yet. The first handler that lands owns introducing the processed_events table + a small shared helper (e.g. already_processed / mark_processed); this section is the convention it must follow.

Why both layers¶

The write-API key prevents client retries from creating duplicates. The handler key prevents redeliveries (re-dispatched failed rows, and broker redeliveries once a module is extracted) from creating duplicates. They protect against different failure modes and you need both.

Logging¶

Levels¶

Format¶

from src.shared.observability import get_logger

logger = get_logger(__name__)

# Good
logger.info(
    "recipe_created",
    recipe_id=str(recipe.id),
    code=recipe.code,
)

# Bad
logger.info(f"Created recipe {recipe.id}")

tenant_id, user_id, trace_id, and span_id are already on the record via structlog.contextvars and the OTel logging instrumentor — don't pass them by hand.

What to Log¶

• Always: tenant_id, user_id, correlation_id
• Business events: entity IDs, status changes
• Never: passwords, tokens, PII

Security¶

Authentication¶

The identity module (not yet shipped) will issue and validate JWTs. Until then, src.shared.auth.current_user.get_current_user is a stub that returns a fixed dev user/tenant pulled from settings. Every request resolves to a CurrentUser(user_id, tenant_id, roles) regardless of which provider is in place.

Forward-compat invariant: when the identity module replaces get_current_user in-place with real JWT validation, endpoint code and the role guards do not change — they only ever read CurrentUser.roles. Token storage, refresh, and API-key issuance are owned by the future identity module.

Authorization¶

Authorization is permission-based. Endpoints declare what capability they gate (e.g. Permission.ITEM_CREATE), never which roles happen to have it today. Role-to-permission policy lives in one mapping; endpoint code is decoupled from it.

Primitives (all in src.shared.auth):

Rules:

• Gate every restricted endpoint with Depends(require_permission(Permission.X)). Do not gate by raw role sets at the endpoint surface; that creates combinatorial alias names and couples endpoints to today's role-to-permission policy.
• RoleName.ADMIN is a wildcard enforced inside the guard. Never list RoleName.ADMIN in PERMISSION_ROLES entries — an empty frozenset() for a permission means "admin only".
• Add the new Permission enum value and its PERMISSION_ROLES entry in the same commit as the endpoint that needs it. The startup check guarantees the build fails immediately if you forget the mapping.
• Permission strings follow "<module>:<resource>:<action>" (e.g. "item_catalog:item:create", "manufacturing:bom:publish") so the registry stays greppable and unambiguous as it grows.
• Tenant isolation is orthogonal and enforced by PostgreSQL RLS, not by role guards. See § Multi-tenancy.

Endpoint pattern:

from typing import Annotated
from fastapi import APIRouter, Depends

from src.shared.auth.current_user import CurrentUser
from src.shared.auth.permissions import Permission
from src.shared.auth.role_guards import require_permission

router = APIRouter(prefix="/items", tags=["item-catalog"])


@router.post("", status_code=201)
async def create_item(
    body: CreateItemRequest,
    repo: ItemRepoDep,
    user: Annotated[CurrentUser, Depends(require_permission(Permission.ITEM_CREATE))],
) -> CreateItemResponse:
    ...

When the role-to-permission policy shifts (e.g. Sales gains item-create rights), only the one line in PERMISSION_ROLES changes — every endpoint that gates on ITEM_CREATE follows automatically.

Resource-ownership / ABAC checks (e.g. "this user can only see items in their tenant", "only the author can edit") are intentionally not in require_permission. RLS handles tenant scoping at the DB layer; per-aggregate ownership belongs in the command/query handler.

The companion /stats endpoint of any list resource shares the list endpoint's read permission — gate /items/stats with the same Permission.ITEM_READ as /items. Do not introduce a separate <RESOURCE>_STATS_READ permission; anyone who can see the list can see its aggregate. See § Stats endpoints.

Input Validation¶

• Pydantic for all inputs
• Validate at API boundary
• Sanitize before storage

OWASP Top 10¶

1. Injection prevention (parameterized queries)
2. Broken auth mitigation (JWT, secure storage)
3. Sensitive data protection (encryption at rest)
4. XXE prevention (disable XML)
5. Broken access control (RBAC, RLS)
6. Security misconfig (secure defaults)
7. XSS prevention (output encoding)
8. Insecure deserialization (strict schemas)
9. Vulnerable dependencies (dependabot)
10. Logging & monitoring (audit trail)

Observability¶

Full reference: docs/architecture/observability.md. Stack: OpenTelemetry → OTel Collector → SigNoz. The rules below are the minimum every module must follow.

What to import¶

from src.shared.observability import counter, get_logger, histogram, start_span, traced_handler

logger = get_logger(__name__)

• get_logger(__name__) — bind a structlog logger named after the current module so each record carries logger=src.modules.<name>... (searchable / filterable in SigNoz). Tenant / user / trace IDs are already in structlog.contextvars, so every call automatically includes them.
• @traced_handler — applied to every Command/Query Handler handle method. Opens a span named {module}.{HandlerClass} and binds module + command/query to the log context.
• counter(name, ...), histogram(name, ...) — OTel metric helpers; declare as module-level singletons.
• start_span(name, **attrs) — context manager for manual spans when a step within a handler is worth measuring on its own.

What to log¶

Do not log on the happy path of Query Handlers — the auto-instrumented HTTP span already records duration and status. Do not log + swallow exceptions; log only at the boundary that decides to suppress.

What to trace¶

• Auto-instrumentation already covers FastAPI requests, SQLAlchemy/asyncpg statements, redis, httpx; the in-process outbox dispatcher emits its own consumer span per event.
• Add a manual span only for:
• A measurable in-process step inside a handler (e.g. multi-level BOM traversal).
• An external boundary that the SDK does not instrument.

@traced_handler
async def handle(self, command: PublishRecipeVersionCommand) -> ...:
    recipe = await self._recipe_repo.get_by_id(command.recipe_id)
    with start_span("manufacturing.calculate_allergens", recipe_id=str(recipe.id)):
        allergens = recipe.calculate_allergens()
    ...

What to meter¶

Default to none. Add a metric only when: - You need an alertable rate or histogram independent of trace sampling. - A dashboard graph needs aggregation across millions of events.

Naming: grinsystem.{module}.{noun}_{unit}. Keep labels low-cardinality — tenant_id is fine; per-entity IDs are not.

Sensitive data¶

Never put passwords, tokens, JWTs, cookies, or Polish national IDs (pesel, nip, regon) into log fields. A structlog processor redacts a configured deny-list, but add new sensitive field names to the deny-list in src/shared/observability/logging.py in the same PR that introduces them — do not rely on call-site discipline.