Difficulty: Advanced
CRUD (Create, Read, Update, Delete) operations are the four basic operations performed on database records. In FastAPI with SQLAlchemy, these operations translate to specific SQLAlchemy methods combined with HTTP endpoints. Organizing CRUD logic into reusable functions keeps your code clean and testable.
The Create operation adds new records to the database. You create an SQLAlchemy model instance, add it to the session with db.add(), commit the transaction with db.commit(), and refresh the object with db.refresh() to load auto-generated fields like the ID. The endpoint typically uses POST and returns 201.
The Read operation retrieves records. SQLAlchemy provides db.query(Model) for building queries. Use .all() to get all records, .first() to get the first match, .filter() to add WHERE clauses, and .get() for primary key lookups. For pagination, use .offset() and .limit(). Read endpoints use GET.
The Update operation modifies existing records. The typical flow is: query the record, modify its attributes, and commit. For partial updates (PATCH), use model_dump(exclude_unset=True) to get only the fields the client sent, then apply them with setattr(). Full updates (PUT) replace all fields.
The Delete operation removes records. Query the record, call db.delete(record), and commit. Return 204 No Content on success. Always check that the record exists before attempting deletion.
A common pattern is to create a CRUD module that encapsulates all database operations for a resource. This separates database logic from HTTP handling, making both easier to test and maintain. The CRUD functions take a session and parameters, perform the database operation, and return the result. The endpoint handler calls the CRUD function and handles HTTP concerns.
from sqlalchemy.orm import Session
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel, ConfigDict
class Base(DeclarativeBase):
pass
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
price: Mapped[float] = mapped_column()
description: Mapped[str | None] = mapped_column(String(500), nullable=True)
class ItemCreate(BaseModel):
name: str
price: float
description: str | None = None
class ItemUpdate(BaseModel):
name: str | None = None
price: float | None = None
description: str | None = None
class ItemResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
price: float
description: str | None
# CRUD functions
def create_item(db: Session, item: ItemCreate) -> Item:
db_item = Item(item.model_dump())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
def get_items(db: Session, skip: int = 0, limit: int = 20) -> list[Item]:
return db.query(Item).offset(skip).limit(limit).all()
def get_item(db: Session, item_id: int) -> Item | None:
return db.query(Item).filter(Item.id == item_id).first()
def update_item(db: Session, item_id: int, item: ItemUpdate) -> Item | None:
db_item = db.query(Item).filter(Item.id == item_id).first()
if not db_item:
return None
update_data = item.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(db_item, key, value)
db.commit()
db.refresh(db_item)
return db_item
def delete_item(db: Session, item_id: int) -> bool:
db_item = db.query(Item).filter(Item.id == item_id).first()
if not db_item:
return False
db.delete(db_item)
db.commit()
return TrueEach CRUD function takes a session and parameters, performs one database operation, and returns the result. create uses add/commit/refresh. get uses query/filter/first. update uses setattr for partial updates. delete uses delete/commit.
from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy.orm import Session
# ... (imports and CRUD functions from above)
app = FastAPI()
@app.post("/items", response_model=ItemResponse, status_code=201)
def create_item_endpoint(item: ItemCreate, db: Session = Depends(get_db)):
return create_item(db, item)
@app.get("/items", response_model=list[ItemResponse])
def list_items_endpoint(
skip: int = Query(default=0, ge=0),
limit: int = Query(default=20, ge=1, le=100),
db: Session = Depends(get_db),
):
return get_items(db, skip=skip, limit=limit)
@app.get("/items/{item_id}", response_model=ItemResponse)
def get_item_endpoint(item_id: int, db: Session = Depends(get_db)):
item = get_item(db, item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
@app.patch("/items/{item_id}", response_model=ItemResponse)
def update_item_endpoint(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)):
updated = update_item(db, item_id, item)
if not updated:
raise HTTPException(status_code=404, detail="Item not found")
return updated
@app.delete("/items/{item_id}", status_code=204)
def delete_item_endpoint(item_id: int, db: Session = Depends(get_db)):
if not delete_item(db, item_id):
raise HTTPException(status_code=404, detail="Item not found")Endpoints are thin wrappers around CRUD functions. They handle HTTP concerns (status codes, error responses, query parameters) while delegating database logic to the CRUD layer. This separation makes both layers independently testable.
from sqlalchemy.orm import Session
from sqlalchemy import or_
def search_items(
db: Session,
q: str | None = None,
min_price: float | None = None,
max_price: float | None = None,
skip: int = 0,
limit: int = 20,
) -> list[Item]:
query = db.query(Item)
if q:
query = query.filter(
or_(
Item.name.ilike(f"%{q}%"),
Item.description.ilike(f"%{q}%"),
)
)
if min_price is not None:
query = query.filter(Item.price >= min_price)
if max_price is not None:
query = query.filter(Item.price <= max_price)
return query.order_by(Item.price).offset(skip).limit(limit).all()
# Usage in endpoint
@app.get("/items/search", response_model=list[ItemResponse])
def search(
q: str | None = None,
min_price: float | None = None,
max_price: float | None = None,
skip: int = 0,
limit: int = 20,
db: Session = Depends(get_db),
):
return search_items(db, q=q, min_price=min_price, max_price=max_price, skip=skip, limit=limit)Build queries incrementally by conditionally adding filters. ilike() performs case-insensitive LIKE matching. or_() combines conditions with OR logic. Pagination is applied last with offset() and limit().
Create, Read, Update, Delete, Query API, Filtering