Python 9 min read

FastAPI Tutorial Part 7: CRUD Operations - Build a Complete REST API

Build production-ready CRUD APIs with FastAPI. Learn RESTful patterns, pagination, filtering, bulk operations, and best practices for real-world applications.

MR

Moshiour Rahman

#fastapi #python #crud #rest-api #tutorial #ai

Advertisement

Understanding CRUD

CRUD represents the four basic operations for persistent storage:

OperationHTTP MethodEndpointDescription
CreatePOST/itemsCreate new resource
ReadGET/items, /items/{id}Retrieve resources
UpdatePUT/PATCH/items/{id}Modify resource
DeleteDELETE/items/{id}Remove resource

Project Structure

app/
├── __init__.py
├── main.py
├── database.py
├── models/
│   ├── __init__.py
│   └── product.py
├── schemas/
│   ├── __init__.py
│   └── product.py
├── crud/
│   ├── __init__.py
│   └── product.py
└── routers/
    ├── __init__.py
    └── products.py

Database Models

# app/models/product.py
from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime, Text
from sqlalchemy.sql import func
from ..database import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(200), nullable=False, index=True)
    slug = Column(String(200), unique=True, index=True)
    description = Column(Text, nullable=True)
    price = Column(Float, nullable=False)
    quantity = Column(Integer, default=0)
    category = Column(String(100), index=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

Pydantic Schemas

# app/schemas/product.py
from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

class ProductBase(BaseModel):
    name: str = Field(min_length=2, max_length=200)
    description: Optional[str] = None
    price: float = Field(gt=0)
    quantity: int = Field(ge=0, default=0)
    category: str = Field(min_length=2, max_length=100)

class ProductCreate(ProductBase):
    slug: str = Field(pattern=r"^[a-z0-9-]+$", min_length=2, max_length=200)

    @field_validator("slug")
    @classmethod
    def lowercase_slug(cls, v: str) -> str:
        return v.lower()

class ProductUpdate(BaseModel):
    name: Optional[str] = Field(default=None, min_length=2, max_length=200)
    description: Optional[str] = None
    price: Optional[float] = Field(default=None, gt=0)
    quantity: Optional[int] = Field(default=None, ge=0)
    category: Optional[str] = Field(default=None, min_length=2, max_length=100)
    is_active: Optional[bool] = None

class ProductResponse(ProductBase):
    id: int
    slug: str
    is_active: bool
    created_at: datetime
    updated_at: Optional[datetime] = None

    class Config:
        from_attributes = True

class ProductList(BaseModel):
    items: List[ProductResponse]
    total: int
    page: int
    per_page: int
    pages: int

class BulkProductCreate(BaseModel):
    products: List[ProductCreate] = Field(min_length=1, max_length=100)

class BulkDeleteRequest(BaseModel):
    ids: List[int] = Field(min_length=1, max_length=100)

class BulkDeleteResponse(BaseModel):
    deleted: int
    failed: List[int]

CRUD Functions

# app/crud/product.py
from sqlalchemy.orm import Session
from sqlalchemy import or_, func
from typing import Optional, List, Tuple
from ..models.product import Product
from ..schemas.product import ProductCreate, ProductUpdate, SortOrder

def get_product(db: Session, product_id: int) -> Optional[Product]:
    return db.query(Product).filter(Product.id == product_id).first()

def get_product_by_slug(db: Session, slug: str) -> Optional[Product]:
    return db.query(Product).filter(Product.slug == slug).first()

def get_products(
    db: Session,
    skip: int = 0,
    limit: int = 20,
    category: Optional[str] = None,
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    search: Optional[str] = None,
    is_active: Optional[bool] = None,
    sort_by: str = "created_at",
    order: SortOrder = SortOrder.desc
) -> Tuple[List[Product], int]:
    query = db.query(Product)

    # Apply filters
    if category:
        query = query.filter(Product.category == category)
    if min_price is not None:
        query = query.filter(Product.price >= min_price)
    if max_price is not None:
        query = query.filter(Product.price <= max_price)
    if search:
        search_term = f"%{search}%"
        query = query.filter(
            or_(
                Product.name.ilike(search_term),
                Product.description.ilike(search_term)
            )
        )
    if is_active is not None:
        query = query.filter(Product.is_active == is_active)

    # Get total before pagination
    total = query.count()

    # Apply sorting
    sort_column = getattr(Product, sort_by, Product.created_at)
    if order == SortOrder.desc:
        query = query.order_by(sort_column.desc())
    else:
        query = query.order_by(sort_column.asc())

    # Apply pagination
    products = query.offset(skip).limit(limit).all()

    return products, total

def create_product(db: Session, product: ProductCreate) -> Product:
    db_product = Product(**product.model_dump())
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

def update_product(
    db: Session,
    product_id: int,
    product: ProductUpdate
) -> Optional[Product]:
    db_product = get_product(db, product_id)
    if not db_product:
        return None

    update_data = product.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_product, field, value)

    db.commit()
    db.refresh(db_product)
    return db_product

def delete_product(db: Session, product_id: int) -> bool:
    db_product = get_product(db, product_id)
    if not db_product:
        return False

    db.delete(db_product)
    db.commit()
    return True

def bulk_create_products(
    db: Session,
    products: List[ProductCreate]
) -> List[Product]:
    db_products = [Product(**p.model_dump()) for p in products]
    db.add_all(db_products)
    db.commit()
    for p in db_products:
        db.refresh(p)
    return db_products

def bulk_delete_products(db: Session, ids: List[int]) -> Tuple[int, List[int]]:
    deleted = 0
    failed = []

    for product_id in ids:
        if delete_product(db, product_id):
            deleted += 1
        else:
            failed.append(product_id)

    return deleted, failed

def get_categories(db: Session) -> List[str]:
    result = db.query(Product.category).distinct().all()
    return [r[0] for r in result if r[0]]

def get_product_stats(db: Session) -> dict:
    return {
        "total_products": db.query(func.count(Product.id)).scalar(),
        "active_products": db.query(func.count(Product.id)).filter(
            Product.is_active == True
        ).scalar(),
        "avg_price": db.query(func.avg(Product.price)).scalar() or 0,
        "total_inventory": db.query(func.sum(Product.quantity)).scalar() or 0
    }

API Router

# app/routers/products.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import Optional, List

from ..database import get_db
from ..crud import product as crud
from ..schemas.product import (
    ProductCreate,
    ProductUpdate,
    ProductResponse,
    ProductList,
    BulkProductCreate,
    BulkDeleteRequest,
    BulkDeleteResponse,
    SortOrder
)

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

# CREATE
@router.post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    """Create a new product."""
    existing = crud.get_product_by_slug(db, product.slug)
    if existing:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Product with this slug already exists"
        )
    return crud.create_product(db, product)

# READ - List with filters
@router.get("/", response_model=ProductList)
def list_products(
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page"),
    category: Optional[str] = Query(None, description="Filter by category"),
    min_price: Optional[float] = Query(None, ge=0, description="Minimum price"),
    max_price: Optional[float] = Query(None, ge=0, description="Maximum price"),
    search: Optional[str] = Query(None, min_length=2, description="Search term"),
    is_active: Optional[bool] = Query(None, description="Filter by active status"),
    sort_by: str = Query("created_at", description="Sort field"),
    order: SortOrder = Query(SortOrder.desc, description="Sort order"),
    db: Session = Depends(get_db)
):
    """List products with filtering, sorting, and pagination."""
    skip = (page - 1) * per_page

    products, total = crud.get_products(
        db,
        skip=skip,
        limit=per_page,
        category=category,
        min_price=min_price,
        max_price=max_price,
        search=search,
        is_active=is_active,
        sort_by=sort_by,
        order=order
    )

    return ProductList(
        items=products,
        total=total,
        page=page,
        per_page=per_page,
        pages=(total + per_page - 1) // per_page
    )

# READ - Single by ID
@router.get("/{product_id}", response_model=ProductResponse)
def get_product(product_id: int, db: Session = Depends(get_db)):
    """Get a product by ID."""
    product = crud.get_product(db, product_id)
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Product not found"
        )
    return product

# READ - Single by slug
@router.get("/slug/{slug}", response_model=ProductResponse)
def get_product_by_slug(slug: str, db: Session = Depends(get_db)):
    """Get a product by slug."""
    product = crud.get_product_by_slug(db, slug)
    if not product:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Product not found"
        )
    return product

# UPDATE - Full update
@router.put("/{product_id}", response_model=ProductResponse)
def update_product(
    product_id: int,
    product: ProductUpdate,
    db: Session = Depends(get_db)
):
    """Update a product (partial update supported)."""
    updated = crud.update_product(db, product_id, product)
    if not updated:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Product not found"
        )
    return updated

# UPDATE - Partial update (same as PUT in this implementation)
@router.patch("/{product_id}", response_model=ProductResponse)
def patch_product(
    product_id: int,
    product: ProductUpdate,
    db: Session = Depends(get_db)
):
    """Partially update a product."""
    return update_product(product_id, product, db)

# DELETE
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(product_id: int, db: Session = Depends(get_db)):
    """Delete a product."""
    if not crud.delete_product(db, product_id):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Product not found"
        )
    return None

# BULK CREATE
@router.post("/bulk", response_model=List[ProductResponse], status_code=status.HTTP_201_CREATED)
def bulk_create_products(
    data: BulkProductCreate,
    db: Session = Depends(get_db)
):
    """Create multiple products at once."""
    # Check for duplicate slugs
    for product in data.products:
        if crud.get_product_by_slug(db, product.slug):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Product with slug '{product.slug}' already exists"
            )

    return crud.bulk_create_products(db, data.products)

# BULK DELETE
@router.delete("/bulk", response_model=BulkDeleteResponse)
def bulk_delete_products(
    data: BulkDeleteRequest,
    db: Session = Depends(get_db)
):
    """Delete multiple products at once."""
    deleted, failed = crud.bulk_delete_products(db, data.ids)
    return BulkDeleteResponse(deleted=deleted, failed=failed)

# METADATA
@router.get("/meta/categories", response_model=List[str])
def get_categories(db: Session = Depends(get_db)):
    """Get all product categories."""
    return crud.get_categories(db)

@router.get("/meta/stats")
def get_stats(db: Session = Depends(get_db)):
    """Get product statistics."""
    return crud.get_product_stats(db)

Main Application

# app/main.py
from fastapi import FastAPI
from .database import engine, Base
from .routers import products

# Create tables
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Product API",
    description="CRUD API for products",
    version="1.0.0"
)

app.include_router(products.router)

@app.get("/")
def root():
    return {"message": "Product API", "docs": "/docs"}

Testing the API

Create Product

curl -X POST "http://localhost:8000/products" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Wireless Mouse",
    "slug": "wireless-mouse",
    "description": "Ergonomic wireless mouse",
    "price": 29.99,
    "quantity": 100,
    "category": "Electronics"
  }'

List with Filters

# Basic list
curl "http://localhost:8000/products"

# With pagination
curl "http://localhost:8000/products?page=2&per_page=10"

# With filters
curl "http://localhost:8000/products?category=Electronics&min_price=10&max_price=50"

# With search
curl "http://localhost:8000/products?search=wireless"

# With sorting
curl "http://localhost:8000/products?sort_by=price&order=asc"

Update Product

curl -X PUT "http://localhost:8000/products/1" \
  -H "Content-Type: application/json" \
  -d '{
    "price": 24.99,
    "quantity": 150
  }'

Delete Product

curl -X DELETE "http://localhost:8000/products/1"

Bulk Operations

# Bulk create
curl -X POST "http://localhost:8000/products/bulk" \
  -H "Content-Type: application/json" \
  -d '{
    "products": [
      {"name": "Product 1", "slug": "product-1", "price": 10, "category": "A"},
      {"name": "Product 2", "slug": "product-2", "price": 20, "category": "B"}
    ]
  }'

# Bulk delete
curl -X DELETE "http://localhost:8000/products/bulk" \
  -H "Content-Type: application/json" \
  -d '{"ids": [1, 2, 3]}'

Advanced Patterns

Soft Delete

# Model
class Product(Base):
    # ... other fields
    deleted_at = Column(DateTime(timezone=True), nullable=True)

# CRUD
def soft_delete_product(db: Session, product_id: int) -> bool:
    product = get_product(db, product_id)
    if not product or product.deleted_at:
        return False

    product.deleted_at = datetime.now(timezone.utc)
    db.commit()
    return True

def get_products(db: Session, include_deleted: bool = False):
    query = db.query(Product)
    if not include_deleted:
        query = query.filter(Product.deleted_at.is_(None))
    return query.all()

Optimistic Locking

# Model
class Product(Base):
    version = Column(Integer, default=1)

# CRUD
def update_product_optimistic(
    db: Session,
    product_id: int,
    product: ProductUpdate,
    expected_version: int
) -> Optional[Product]:
    db_product = get_product(db, product_id)
    if not db_product:
        return None

    if db_product.version != expected_version:
        raise HTTPException(
            status_code=409,
            detail="Resource has been modified"
        )

    update_data = product.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_product, field, value)

    db_product.version += 1
    db.commit()
    db.refresh(db_product)
    return db_product

Summary

OperationMethodEndpointStatus
CreatePOST/products201
ListGET/products200
Get OneGET/products/{id}200
UpdatePUT/PATCH/products/{id}200
DeleteDELETE/products/{id}204
Bulk CreatePOST/products/bulk201
Bulk DeleteDELETE/products/bulk200
PatternUse Case
PaginationLarge datasets
FilteringSearch and filter
SortingOrder results
Soft DeletePreserve history
Optimistic LockingConcurrent updates

Next Steps

In Part 8, we’ll explore Error Handling and Exceptions - creating custom error responses and handling various failure scenarios gracefully.

Series Navigation:

  • Part 1-6: Foundations
  • Part 7: CRUD Operations (You are here)
  • Part 8: Error Handling

Advertisement

MR

Moshiour Rahman

Software Architect & AI Engineer

Share:
MR

Moshiour Rahman

Software Architect & AI Engineer

Enterprise software architect with deep expertise in financial systems, distributed architecture, and AI-powered applications. Building large-scale systems at Fortune 500 companies. Specializing in LLM orchestration, multi-agent systems, and cloud-native solutions. I share battle-tested patterns from real enterprise projects.

Related Articles

Python

FastAPI Tutorial Part 1: Introduction and Setup - Build Modern Python APIs

Start your FastAPI journey with this comprehensive guide. Learn installation, create your first API, understand async Python, and explore automatic documentation.

Python

FastAPI Tutorial Part 2: Path and Query Parameters - Complete Guide

Master FastAPI path and query parameters. Learn parameter validation, type conversion, optional parameters, and advanced patterns for building flexible APIs.

Python

FastAPI Tutorial Part 4: Response Models and Status Codes - Control Your API Output

Learn to control FastAPI responses with response models, status codes, and headers. Master data filtering, multiple response types, and proper HTTP semantics.

Comments

Comments are powered by GitHub Discussions.

Configure Giscus at giscus.app to enable comments.