flowCreate.solutions

Generic Example — CRUD

This page shows the CRUD layer (async SQLAlchemy database operations). For the full end-to-end example (including tenant access checks at the API layer), see:

What belongs in crud.py

  • Database interactions (create/get/list/update/delete)
  • Query composition, eager loading, ordering
  • Transaction boundaries (commit, refresh)

What does NOT belong here

  • Authentication & authorization (that belongs in apis.py / dependencies)
  • HTTP response formatting (keep that in the API layer)
  • Response envelopes (StandardResponse / StandardListResponse) — CRUD should return domain objects/primitives

Example crud.py

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()

Notes

  • Tenant scoping should be enforced by requiring tenant_id in CRUD signatures for tenant-owned resources.
  • Use exclude_unset=True for updates so you don’t overwrite fields with null unintentionally.