The Full CRUD Pattern

A well-structured CRUD router follows a consistent pattern: Pydantic schema → async DB call → response model. Let’s build a complete articles resource.

┌──────────┐    ┌────────────┐    ┌──────────┐    ┌──────────┐
│  Client   │───►│  Pydantic  │───►│  AsyncDB │───►│  Schema  │───► JSON
│  Request  │    │  Validate  │    │  Execute │    │  Filter  │
└──────────┘    └────────────┘    └──────────┘    └──────────┘
                     │                  │
                     ▼                  ▼
               422 Validation     500/404 Error

Project Structure

app/
├── models.py        # SQLAlchemy models
├── schemas.py       # Pydantic schemas (Create, Update, Read)
├── database.py      # Engine, session, Base
├── crud.py          # Pure DB functions (optional layer)
└── routers/
    └── articles.py  # FastAPI router

SQLAlchemy Model

# app/models.py
from sqlalchemy import String, Integer, Boolean, DateTime, Text, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
from datetime import datetime


class Article(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    content: Mapped[str] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(Boolean, default=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())

    author: Mapped["User"] = relationship(back_populates="articles")

Pydantic Schemas

Separate schemas for Create, Update, and Read - the three-schema pattern:

# app/schemas.py
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import Optional


# ── Article Schemas ──

class ArticleCreate(BaseModel):
    title: str
    content: str
    published: bool = False
    author_id: int


class ArticleUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None
    published: Optional[bool] = None


class ArticleRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    content: str
    published: bool
    author_id: int
    created_at: datetime
    updated_at: datetime


# ── User Summary (nested) ──

class UserSummary(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    email: str


class ArticleWithAuthor(ArticleRead):
    author: UserSummary

Schema	fields_required	Used For
ArticleCreate	All fields (with defaults for optional)	POST /articles
ArticleUpdate	All fields Optional	PATCH /articles/{id}
ArticleRead	All fields, from_attributes=True	response_model

CRUD Router

# app/routers/articles.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy import select, func, delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.models import Article
from app.schemas import ArticleCreate, ArticleUpdate, ArticleRead, ArticleWithAuthor
from typing import Optional

router = APIRouter(prefix="/articles", tags=["articles"])


# ── CREATE ──

@router.post("/", response_model=ArticleRead, status_code=status.HTTP_201_CREATED)
async def create_article(payload: ArticleCreate, db: AsyncSession = Depends(get_db)):
    article = Article(**payload.model_dump())
    db.add(article)
    await db.commit()
    await db.refresh(article)
    return article

curl -X POST http://localhost:8000/articles \
  -H "Content-Type: application/json" \
  -d '{"title": "Async FastAPI", "content": "Deep dive...", "author_id": 1}'

# 201 Created
# {"id":1,"title":"Async FastAPI","content":"Deep dive...","published":false,...}

READ - Single

@router.get("/{article_id}", response_model=ArticleWithAuthor)
async def get_article(article_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Article).where(Article.id == article_id)
    )
    article = result.scalar_one_or_none()
    if not article:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Article {article_id} not found",
        )
    return article

READ - List with Pagination

@router.get("/", response_model=dict)
async def list_articles(
    skip: int = Query(0, ge=0, description="Records to skip"),
    limit: int = Query(10, ge=1, le=100, description="Page size"),
    published: Optional[bool] = None,
    db: AsyncSession = Depends(get_db),
):
    query = select(Article)

    if published is not None:
        query = query.where(Article.published == published)

    # Get total count
    count_query = select(func.count()).select_from(query.subquery())
    total = await db.scalar(count_query)

    # Get page
    query = query.order_by(Article.created_at.desc()).offset(skip).limit(limit)
    result = await db.execute(query)
    articles = result.scalars().all()

    return {
        "items": articles,
        "total": total,
        "skip": skip,
        "limit": limit,
    }

curl "http://localhost:8000/articles?skip=0&limit=5&published=true"

# 200 OK
# {
#   "items": [...],
#   "total": 42,
#   "skip": 0,
#   "limit": 5
# }

UPDATE - Full Replacement (PUT)

@router.put("/{article_id}", response_model=ArticleRead)
async def replace_article(article_id: int, payload: ArticleCreate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Article).where(Article.id == article_id))
    article = result.scalar_one_or_none()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")

    for key, value in payload.model_dump().items():
        setattr(article, key, value)

    await db.commit()
    await db.refresh(article)
    return article

UPDATE - Partial (PATCH)

@router.patch("/{article_id}", response_model=ArticleRead)
async def patch_article(article_id: int, payload: ArticleUpdate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Article).where(Article.id == article_id))
    article = result.scalar_one_or_none()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")

    update_data = payload.model_dump(exclude_unset=True)
    if not update_data:
        raise HTTPException(status_code=400, detail="No fields to update")

    for key, value in update_data.items():
        setattr(article, key, value)

    await db.commit()
    await db.refresh(article)
    return article

exclude_unset=True ensures only the fields the client actually sent are applied. This prevents accidentally setting fields to None.

DELETE

@router.delete("/{article_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_article(article_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Article).where(Article.id == article_id))
    article = result.scalar_one_or_none()
    if not article:
        raise HTTPException(status_code=404, detail="Article not found")

    await db.delete(article)
    await db.commit()

curl -X DELETE http://localhost:8000/articles/1
# 204 No Content

curl -X DELETE http://localhost:8000/articles/999
# 404 Not Found
# {"detail":"Article 999 not found"}

Status Code Quick Reference

Operation	Success Code	Error Codes
POST (create)	201 Created	422 validation, 409 conflict
GET (single)	200 OK	404 not found, 422 bad ID
GET (list)	200 OK	422 invalid query params
PUT (replace)	200 OK	404 not found, 422 validation
PATCH (partial)	200 OK	404 not found, 400 no fields
DELETE	204 No Content	404 not found

Error Handling

Custom Error Responses

from fastapi.responses import JSONResponse

class NotFoundError(HTTPException):
    def __init__(self, resource: str, resource_id):
        super().__init__(
            status_code=404,
            detail={
                "error": "not_found",
                "resource": resource,
                "id": str(resource_id),
                "message": f"{resource} with id {resource_id} does not exist",
            },
        )

@router.get("/{article_id}")
async def get_article(article_id: int, db: AsyncSession = Depends(get_db)):
    ...
    if not article:
        raise NotFoundError("Article", article_id)

Global Exception Handler

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "status": "error",
            "code": exc.status_code,
            "message": exc.detail,
            "path": request.url.path,
        },
    )

Database Integrity Errors

from sqlalchemy.exc import IntegrityError

@router.post("/", response_model=ArticleRead, status_code=201)
async def create_article(payload: ArticleCreate, db: AsyncSession = Depends(get_db)):
    try:
        article = Article(**payload.model_dump())
        db.add(article)
        await db.commit()
        await db.refresh(article)
        return article
    except IntegrityError as e:
        await db.rollback()
        raise HTTPException(
            status_code=409,
            detail="Resource already exists or foreign key violation",
        )

Pagination Models

For reusable pagination, define a generic Pydantic schema:

from pydantic import BaseModel, Field
from typing import Generic, TypeVar

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int = Field(..., description="Total number of items")
    skip: int = Field(..., description="Number of items skipped")
    limit: int = Field(..., description="Items per page")
    pages: int = Field(..., description="Total number of pages")

    @classmethod
    def create(cls, items: list[T], total: int, skip: int, limit: int):
        return cls(
            items=items,
            total=total,
            skip=skip,
            limit=limit,
            pages=max(1, (total + limit - 1) // limit),
        )


@router.get("/", response_model=PaginatedResponse[ArticleRead])
async def list_articles(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    # Count
    total = await db.scalar(select(func.count(Article.id)))

    # Fetch page
    result = await db.execute(
        select(Article).order_by(Article.created_at.desc()).offset(skip).limit(limit)
    )
    articles = result.scalars().all()

    return PaginatedResponse.create(articles, total, skip, limit)

Connecting the Router

# app/main.py
from fastapi import FastAPI
from app.routers import articles

app = FastAPI()
app.include_router(articles.router, prefix="/api/v1")

GET  /api/v1/articles        # List (paginated)
GET  /api/v1/articles/42     # Single
POST /api/v1/articles        # Create
PUT  /api/v1/articles/42     # Full replace
PATCH /api/v1/articles/42    # Partial update
DELETE /api/v1/articles/42   # Delete

Key Takeaways

• Use the three-schema pattern - Create, Update, and Read Pydantic schemas - to separate input validation from output serialization
• Set response_model on every endpoint; it filters fields, generates OpenAPI docs, and enables serialization
• Use correct status codes: 201 for POST, 204 for DELETE, 200 for everything else
• Paginate list endpoints with offset/limit (or cursor-based); always return total count
• Raise HTTPException for error paths rather than returning error dicts manually
• Catch IntegrityError from SQLAlchemy to return meaningful 409 Conflict responses
• Use exclude_unset=True on PATCH requests to avoid overwriting fields the client didn’t send
• Register routers with app.include_router() and version your API prefix (/api/v1)