Generic Example — End-to-End Module Walkthrough

This walkthrough shows a complete, generic database module following our standards layout:

database/items/
├── __init__.py
├── models.py
├── schemas.py
├── crud.py
├── apis.py
└── docs/
    └── readme.md

The example is intentionally generic:

It uses tenant/company scoping as the example multi-tenant boundary.
It uses max lengths on all user-provided strings, and reserves HTML sanitization for explicit rich-text fields only.
It keeps CRUD responsible for DB interactions, and API handlers responsible for auth + HTTP responses.

If your project uses different names (tenant_id vs company_id), keep the pattern and rename consistently.

1) `models.py` — SQLAlchemy models

Rules

Prefer explicit constraints (nullable, index, ForeignKey).
Include created_at / updated_at timestamps.
Add tenant scoping fields (tenant_id) to any tenant-owned table and index them.

from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, func, Index
from database.database import Base


class Item(Base):
    __tablename__ = "items"

    id = Column(String, primary_key=True, index=True)

    # Multi-tenant boundary: every item belongs to exactly one tenant.
    tenant_id = Column(String, ForeignKey("tenants.id"), nullable=False, index=True)

    name = Column(String, nullable=False)
    description = Column(String, nullable=True)
    price_cents = Column(Integer, nullable=False)

    created_at = Column(DateTime, server_default=func.now(), nullable=False)
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)


# Example composite index to support common query patterns:
Index("ix_items_tenant_id_created_at", Item.tenant_id, Item.created_at)

2) `schemas.py` — Pydantic validation & serialization

Rules

Every string input must have a max_length.
Sanitize only fields intended to store/render HTML (rich text).
Use separate schemas for create, update, and response.
For updates, use exclude_unset=True so PATCH-like behaviour is consistent.

from __future__ import annotations

from datetime import datetime
from typing import Optional

from pydantic import BaseModel, ConfigDict, Field, field_validator


class ItemCreate(BaseModel):
    name: str = Field(..., max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price_cents: int = Field(..., ge=1, le=10_000_000)


class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    price_cents: Optional[int] = Field(None, ge=1, le=10_000_000)


class ItemResponse(BaseModel):
    id: str
    tenant_id: str
    name: str
    description: Optional[str]
    price_cents: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)

3) `crud.py` — Database operations (async SQLAlchemy)

Rules

CRUD functions should require tenant_id for tenant-scoped resources.
Prefer returning ORM objects (API layer decides the response envelope).
Commit/refresh consistently on mutations.

from __future__ import annotations

from typing import Optional, Sequence
from uuid import uuid4

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from .models import Item
from .schemas import ItemCreate, ItemUpdate


def _new_id() -> str:
    return str(uuid4())


async def create_item(db: AsyncSession, tenant_id: str, payload: ItemCreate) -> Item:
    item = Item(id=_new_id(), tenant_id=tenant_id, **payload.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item


async def get_item(db: AsyncSession, tenant_id: str, item_id: str) -> Optional[Item]:
    res = await db.execute(select(Item).where(Item.tenant_id == tenant_id, Item.id == item_id))
    return res.scalar_one_or_none()


async def list_items(db: AsyncSession, tenant_id: str, page: int, page_size: int) -> Sequence[Item]:
    offset = (page - 1) * page_size
    res = await db.execute(
        select(Item)
        .where(Item.tenant_id == tenant_id)
        .offset(offset)
        .limit(page_size)
    )
    return res.scalars().all()


async def update_item(db: AsyncSession, item: Item, payload: ItemUpdate) -> Item:
    update_data = payload.model_dump(exclude_unset=True)
    for k, v in update_data.items():
        setattr(item, k, v)
    await db.commit()
    await db.refresh(item)
    return item


async def delete_item(db: AsyncSession, item: Item) -> None:
    await db.delete(item)
    await db.commit()

4) `apis.py` — FastAPI routing + authz + response envelopes

Rules

API handlers enforce authentication + tenant access checks.
Use consistent response models (envelopes) across modules.
Avoid leaking cross-tenant data: always scope by tenant and verify access.

from __future__ import annotations

from fastapi import APIRouter, Depends, Header, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from database.database import get_db
from database.schemas import StandardListResponse, StandardResponse
from database.dependencies import get_current_user

from . import crud
from .schemas import ItemCreate, ItemResponse, ItemUpdate


async def require_tenant_access(db: AsyncSession, user_id: str, tenant_id: str) -> None:
    """
    Replace this with your project’s multi-tenant authorization rule.
    Must raise 403 if the user is not allowed to access the tenant.
    """
    allowed = True  # placeholder
    if not allowed:
        raise HTTPException(status_code=403, detail="Access denied")


router = APIRouter(prefix="/api/v1/items", tags=["Items"])


async def get_tenant_id_from_header_or_session(
    token: dict,
    x_tenant_id: str | None = Header(default=None, alias="X-Tenant-Id"),
) -> str:
    """
    Tenant selection standard:
    - Header-selected mode: tenant comes from `X-Tenant-Id` (exact casing).
    - Derived-tenant mode: tenant comes from the authenticated session/JWT active tenant.
    - These modes are mutually exclusive; if the wrong tenant signal is present, reject (client error).

    NOTE: This is a documentation example; implement this in a shared dependency in real projects.
    """
    active_tenant_id = token.get("active_tenant_id")

    # Derived-tenant mode (active tenant session)
    if active_tenant_id:
        if x_tenant_id:
            raise HTTPException(status_code=400, detail="Do not send X-Tenant-Id in derived-tenant mode")
        return str(active_tenant_id)

    # Header-selected mode (multi-tenant switching)
    if not x_tenant_id:
        raise HTTPException(status_code=400, detail="Missing X-Tenant-Id")
    return x_tenant_id


@router.post("", response_model=StandardResponse)
async def create_item(
    payload: ItemCreate,
    db: AsyncSession = Depends(get_db),
    token: dict = Depends(get_current_user),
    tenant_id: str = Depends(get_tenant_id_from_header_or_session),
):
    await require_tenant_access(db, token["user_id"], tenant_id)
    item = await crud.create_item(db, tenant_id=tenant_id, payload=payload)
    return StandardResponse(
        success=True,
        title="Item created",
        content={"item": ItemResponse.model_validate(item).model_dump()},
    )


@router.get("/{item_id}", response_model=StandardResponse)
async def get_item(
    item_id: str,
    db: AsyncSession = Depends(get_db),
    token: dict = Depends(get_current_user),
    tenant_id: str = Depends(get_tenant_id_from_header_or_session),
):
    await require_tenant_access(db, token["user_id"], tenant_id)
    item = await crud.get_item(db, tenant_id=tenant_id, item_id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Not found")
    return StandardResponse(
        success=True,
        title="Item retrieved",
        content={"item": ItemResponse.model_validate(item).model_dump()},
    )


@router.get("", response_model=StandardListResponse)
async def list_items(
    page: int = 1,
    page_size: int = 50,
    db: AsyncSession = Depends(get_db),
    token: dict = Depends(get_current_user),
    tenant_id: str = Depends(get_tenant_id_from_header_or_session),
):
    await require_tenant_access(db, token["user_id"], tenant_id)
    items = await crud.list_items(db, tenant_id=tenant_id, page=page, page_size=page_size)
    return StandardListResponse(
        success=True,
        title="Items retrieved",
        content=[ItemResponse.model_validate(i).model_dump() for i in items],
    )


@router.patch("/{item_id}", response_model=StandardResponse)
async def update_item(
    item_id: str,
    payload: ItemUpdate,
    db: AsyncSession = Depends(get_db),
    token: dict = Depends(get_current_user),
    tenant_id: str = Depends(get_tenant_id_from_header_or_session),
):
    await require_tenant_access(db, token["user_id"], tenant_id)
    item = await crud.get_item(db, tenant_id=tenant_id, item_id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Not found")
    updated = await crud.update_item(db, item=item, payload=payload)
    return StandardResponse(
        success=True,
        title="Item updated",
        content={"item": ItemResponse.model_validate(updated).model_dump()},
    )


@router.delete("/{item_id}", response_model=StandardResponse)
async def delete_item(
    item_id: str,
    db: AsyncSession = Depends(get_db),
    token: dict = Depends(get_current_user),
    tenant_id: str = Depends(get_tenant_id_from_header_or_session),
):
    await require_tenant_access(db, token["user_id"], tenant_id)
    item = await crud.get_item(db, tenant_id=tenant_id, item_id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Not found")
    await crud.delete_item(db, item=item)
    return StandardResponse(success=True, title="Item deleted", content={"id": item_id})

5) `docs/readme.md` — Module documentation

In the reference backend, every module has a consistent docs/readme.md structure. Your standards should encourage the same: summary, data model, schemas/validation, endpoints, workflows, integrations, security, and testing notes.

Example skeleton:

# Items Module

## Summary
- What this module does and why it exists.

## Data Model
### Primary Tables
- `items`: ...

## Schemas & Validation
- `ItemCreate`, `ItemUpdate`, `ItemResponse`: ...

## API Endpoints
- `POST /api/v1/items` (with `X-Tenant-Id` OR derived-tenant mode): ...
- `GET /api/v1/items?page=1&page_size=50` (with `X-Tenant-Id` OR derived-tenant mode): ...

## Workflows & Business Logic
- ...

## Security & Compliance
- Sanitization, authz checks, rate limits, tenant isolation.

## Testing & Monitoring (optional)
- Link to `tests/items/`.

Common pitfalls checklist

Tenant isolation: every query must scope by tenant/company.
Validation: every string input has max lengths; only rich-text fields are HTML-sanitized.
Update semantics: use exclude_unset=True for patch-like updates.
Errors: keep API error handling consistent (404 vs 403, etc.).
Don’t leak internals: response envelopes should hide internal fields not meant for clients.

Generic Example — End-to-End Module Walkthrough

1) models.py — SQLAlchemy models

2) schemas.py — Pydantic validation & serialization

3) crud.py — Database operations (async SQLAlchemy)

4) apis.py — FastAPI routing + authz + response envelopes

5) docs/readme.md — Module documentation

Common pitfalls checklist

1) `models.py` — SQLAlchemy models

2) `schemas.py` — Pydantic validation & serialization

3) `crud.py` — Database operations (async SQLAlchemy)

4) `apis.py` — FastAPI routing + authz + response envelopes

5) `docs/readme.md` — Module documentation