flowCreate.solutions

Generic Example — APIs

This page shows the API layer (FastAPI routing) for a typical entity. For the full end-to-end pattern, see:

What belongs in apis.py

  • Route declarations and versioned paths
  • Dependency injection (db, current_user)
  • AuthZ checks (tenant/company access enforcement)
  • Response envelopes (consistent success/error shapes)

Example apis.py

from __future__ import annotations

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

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

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

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


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


@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 = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    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],
    )

Notes

  • Keep tenant scoping consistent across your tenant selection dependency and CRUD function signatures.
  • Standardize tenant switching via X-Tenant-Id (exact casing) for multi-tenant switching.
  • Standardize derived-tenant mode for “active tenant sessions” (tenant comes from session/JWT).
  • Treat these two modes as mutually exclusive and reject ambiguous requests.
  • Only return fields clients should see (use response schemas to control exposure).