The challenge of API structure
By day 77 of backend development, you've likely built several APIs but found yourself wrestling with the same question: where does the logic actually live? Building a CRUD API with FastAPI and SQLAlchemy explained forces us to think in layers, and getting this separation right determines whether your codebase scales or becomes a tangled mess of domain logic scattered across endpoints.
In this guide, we'll walk through the complete anatomy of a production-grade CRUD API: from the database model through Pydantic schemas, into a service layer that holds the business rules, and finally the router that glues it all together. We'll use a product catalog as our running example because inventory management appears in e-commerce, ride-hailing, and SaaS platforms alike.
SQLAlchemy models: Your database contract
The model is the foundation. It defines the table structure, constraints, and relationships. In production systems, this is where you enforce data integrity at the database level, not just in application code.
Here's a realistic product model with audit fields and indexes:
from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean, Index
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime, timezone
Base = declarative_base()
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
sku = Column(String(50), unique=True, nullable=False, index=True)
name = Column(String(255), nullable=False)
description = Column(String(2000))
price = Column(Float, nullable=False)
cost = Column(Float, nullable=False)
stock = Column(Integer, default=0, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), nullable=False)
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc), nullable=False)
__table_args__ = (
Index('idx_products_sku_active', 'sku', 'is_active'),
)
Notice the composite index on SKU and active status—this reflects how you'll actually query products in a real platform. The updated_at field with onupdate means the database handles the timestamp, not your application. This prevents time-sync bugs across servers.
Pydantic schemas: The API contract
Schemas define what clients send and what they receive. Separating Create, Update, and Out models gives you flexibility: the Out model might expose fields (like timestamps) that clients shouldn't send, while Create and Update models define what's actually required.
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional
class ProductCreate(BaseModel):
sku: str = Field(..., min_length=1, max_length=50)
name: str = Field(..., min_length=1, max_length=255)
description: Optional[str] = Field(None, max_length=2000)
price: float = Field(..., gt=0)
cost: float = Field(..., ge=0)
stock: int = Field(default=0, ge=0)
@field_validator('sku')
@classmethod
def sku_uppercase(cls, v):
return v.upper().strip()
class ProductUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=1, max_length=255)
description: Optional[str] = Field(None, max_length=2000)
price: Optional[float] = Field(None, gt=0)
cost: Optional[float] = Field(None, ge=0)
stock: Optional[int] = Field(None, ge=0)
is_active: Optional[bool] = None
class ProductOut(BaseModel):
id: int
sku: str
name: str
description: Optional[str]
price: float
cost: float
stock: int
is_active: bool
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
The from_attributes=True (formerly orm_mode) tells Pydantic to read from SQLAlchemy ORM objects directly. The validators ensure data cleanliness before it hits the database. This is where you catch malformed requests early, not in your business logic.
Service layer: Where business rules live
The service layer is the crucial middle ground. It handles queries, enforces business rules, and transforms data between the ORM and schemas. Think of it as the "why" of your API—why you check stock before accepting an order, or why you prevent duplicate SKUs.
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException, status
class ProductService:
@staticmethod
def create(db: Session, data: ProductCreate) -> Product:
"""Create a new product. Raises 409 if SKU already exists."""
try:
product = Product(
sku=data.sku,
name=data.name,
description=data.description,
price=data.price,
cost=data.cost,
stock=data.stock,
)
db.add(product)
db.commit()
db.refresh(product)
return product
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Product with SKU '{data.sku}' already exists"
)
@staticmethod
def get_by_id(db: Session, product_id: int) -> Product:
"""Fetch a product by ID. Returns None if not found."""
return db.query(Product).filter(Product.id == product_id).first()
@staticmethod
def list_active(db: Session, skip: int = 0, limit: int = 100):
"""List active products with pagination."""
return db.query(Product).filter(Product.is_active == True).offset(skip).limit(limit).all()
@staticmethod
def update(db: Session, product_id: int, data: ProductUpdate) -> Product:
"""Update a product. Only provided fields are changed."""
product = ProductService.get_by_id(db, product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(product, key, value)
try:
db.commit()
db.refresh(product)
return product
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Could not update product. Check for duplicate SKU."
)
@staticmethod
def delete(db: Session, product_id: int) -> bool:
"""Soft-delete a product by marking inactive."""
product = ProductService.get_by_id(db, product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
product.is_active = False
db.commit()
return True
Notice we're catching IntegrityError and translating it to a meaningful HTTP 409. We use exclude_unset=True on the update to only change fields the client explicitly provided—this prevents accidentally nulling fields. The delete is a soft delete via is_active, a common pattern in platforms where audit trails matter.
Router: Assembling the endpoints
The router connects schemas to services and defines the HTTP contract. Each endpoint is thin: it extracts dependencies, calls the service, and returns the schema.
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import List
router = APIRouter(prefix="/products", tags=["products"])
def get_db():
"""Dependency for database session."""
db = SessionLocal()
try:
yield db
finally:
db.close()
products_service = ProductService()
@router.post("/", response_model=ProductOut, status_code=201)
def create(data: ProductCreate, db: Session = Depends(get_db)):
"""Create a new product."""
return products_service.create(db, data)
@router.get("/{product_id}", response_model=ProductOut)
def get(product_id: int, db: Session = Depends(get_db)):
"""Fetch a single product by ID."""
product = products_service.get_by_id(db, product_id)
if not product:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Product not found"
)
return product
@router.get("/", response_model=List[ProductOut])
def list_products(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=500),
db: Session = Depends(get_db)
):
"""List active products with pagination."""
return products_service.list_active(db, skip=skip, limit=limit)
@router.patch("/{product_id}", response_model=ProductOut)
def update(
product_id: int,
data: ProductUpdate,
db: Session = Depends(get_db)
):
"""Update a product (partial update via PATCH)."""
return products_service.update(db, product_id, data)
@router.delete("/{product_id}", status_code=204)
def delete(product_id: int, db: Session = Depends(get_db)):
"""Soft-delete a product."""
products_service.delete(db, product_id)
return None
The endpoints are readable because the service handles the logic. Status codes are explicit: 201 for creation, 204 for successful deletes with no content. The get_db dependency ensures each request has its own session and cleanup happens automatically.
Wiring it into your FastAPI app
In your main application file:
from fastapi import FastAPI
from app.routers import products
app = FastAPI(title="Product API", version="1.0.0")
app.include_router(products.router)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Common mistakes
- Mixing business logic and HTTP logic: Putting query filters or validation directly in the router makes it hard to test and reuse logic across different endpoints or async tasks. Move it to the service layer.
- Forgetting the Out schema: Returning raw ORM objects exposes internal structure and makes it harder to evolve your database without breaking clients. Always serialize through Pydantic.
- Not handling database errors:
IntegrityError,OperationalError, and timeout exceptions will crash your API if uncaught. Catch them in the service and return appropriate HTTP status codes. - Assuming all updates are partial: Using PATCH (partial) vs PUT (full replacement) matters. Be consistent and document which endpoint expects which behavior.
- Creating a new service instance per request: Make your service class static or use a single instance. Creating new ones wastes memory and breaks any internal caching you might add.
Key takeaways
- SQLAlchemy models define your database contract; use constraints and indexes to enforce integrity at the source.
- Pydantic schemas separate concerns: Create for input, Update for partial changes, Out for API responses. This flexibility pays off in production.
- The service layer is where business rules live—queries, validations, error handling. Keep it independent of HTTP to enable reuse and testing.
- Routers are thin and focused. They wire dependencies, call services, and return schemas. If a router method is complex, that logic belongs in the service.
- Use appropriate HTTP status codes and handle database exceptions gracefully. Your API consumers depend on predictable error responses.
- Soft deletes via an
is_activeflag are common in platforms requiring audit trails; consider this early in your schema design.
Comments
Loading comments…