Search Topics
Search across all FastAPI topics
CRUD Operations
Core ConceptCreate, Read, Update, Delete. Four operations that power every API. Get the patterns right, or one missing filter deletes your entire database.
You write a DELETE endpoint: db.query(User).delete(). No filter. You just wiped every user in your production database. The commit was automatic because autoflush was on.
# Your "delete user" endpoint:
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
db.query(User).delete() # <- WHERE IS THE FILTER?!
db.commit()
return {"status": "deleted"}
# What you meant: delete user #42
# What you did: DELETE FROM users (ALL of them)
# Rows deleted: 15,847
# Time to realize: 3 minutes
# Time to recover from backup: 4 hoursQuestion
Scary, right? The endpoint receives user_id as a parameter but never uses it. SQLAlchemy doesn't warn you about unfiltered deletes — it just does what you ask. That's why the safest pattern is: fetch first, then delete. If you can't find it, you can't accidentally delete everything.
The repository pattern: keeping your routes clean
Put all database logic in a crud.py file. Your route handlers should only handle HTTP concerns — status codes, response models, error responses. The CRUD functions handle the database.
Route Handler
HTTP request/response
CRUD Function
Database logic
SQLAlchemy Session
Execute SQL
Database
Store/retrieve data
Architecture
What you just learned
Separate database logic (crud.py) from HTTP logic (routes)
CRUD functions take a Session and return data — nothing HTTP-related
This makes your database logic testable without spinning up a server
Create: adding new records
Convert a Pydantic schema to an ORM object, add it to the session, commit, and refresh. That last step is important — refresh() pulls back the database-generated fields like id.
# crud.py
from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate
def create_user(db: Session, user: UserCreate) -> User:
db_user = User(**user.model_dump()) # Pydantic -> dict -> ORM object
db.add(db_user) # Stage the object
db.commit() # Write to database
db.refresh(db_user) # Pull back the generated id
return db_user
# main.py — the route handler stays simple
@app.post("/users", response_model=UserOut)
def create(user: UserCreate, db: Session = Depends(get_db)):
return create_user(db, user)Insight
Why model_dump() instead of passing the Pydantic object directly? Because User(**user.model_dump()) unpacks the dict as keyword arguments. It's like writing User(name="Alice", email="alice@example.com"). The ORM constructor expects keyword args, not a Pydantic object.
Read: fetching records
Three patterns you'll use constantly: get by ID, get a paginated list, and filtered queries. Notice how every "get by ID" can return None — you need to handle that in the route handler.
# Single record by ID — returns None if not found
def get_user(db: Session, user_id: int) -> User | None:
return db.query(User).filter(User.id == user_id).first()
# Paginated list — skip and limit prevent loading 10,000 rows
def get_users(
db: Session,
skip: int = 0,
limit: int = 100,
) -> list[User]:
return db.query(User).offset(skip).limit(limit).all()
# Filtered query — add as many .filter() calls as you need
def get_active_users(db: Session) -> list[User]:
return db.query(User).filter(User.is_active == True).all()
# main.py — always check for None!
@app.get("/users/{user_id}", response_model=UserOut)
def read_user(user_id: int, db: Session = Depends(get_db)):
user = get_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return userRead Operations
What you just learned
first() returns None for empty results — it never raises an exception
Always add skip/limit to list queries. Loading all rows kills performance.
The 404 check lives in the route handler, not in the CRUD function
Update: partial vs. full updates
For partial updates, use exclude_unset=True. This is the key — it only modifies fields the client actually sent, not every field in the schema. Without it, omitted fields get set to their default (usually None), which can wipe out existing data.
from schemas import UserUpdate # All fields are Optional
def update_user(
db: Session,
user_id: int,
updates: UserUpdate,
) -> User | None:
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return None
# The important part: only update fields that were explicitly set
update_data = updates.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_user, field, value)
db.commit()
db.refresh(db_user)
return db_user
# main.py
@app.patch("/users/{user_id}", response_model=UserOut)
def update(user_id: int, updates: UserUpdate, db: Session = Depends(get_db)):
user = update_user(db, user_id, updates)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return userDelete: fetch first, then delete
The safe pattern: find the record first, then delete it. If it doesn't exist, return None (and the route handler raises 404). Never use an unfiltered .delete() — that's how the mass deletion from the hook happens.
def delete_user(db: Session, user_id: int) -> User | None:
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
return None
db.delete(db_user) # Delete THIS specific object
db.commit()
return db_user
# Alternative: soft delete (mark as inactive, don't actually remove)
# db_user.is_active = False
# db.commit()
# return db_user
# main.py
@app.delete("/users/{user_id}", response_model=UserOut)
def delete(user_id: int, db: Session = Depends(get_db)):
user = delete_user(db, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return userWatch out
Consider soft deletes for anything users create. is_active = False is a lot easier to undo than DELETE FROM users WHERE id = 42. Hard deletes are permanent. Backups take hours to restore. Soft deletes take one UPDATE to reverse.
Delete Operations
What you just learned
Always fetch before delete — db.delete(object) is safer than .delete() on a query
Soft deletes (is_active=False) are reversible. Hard deletes aren't.
The CRUD function returns None for 'not found'. The route handler raises 404.
Go Deeper: The unfiltered delete
Mass deletion without a filter
You copy-paste a query pattern and forget to add the filter. The delete runs against the entire table.
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
# You have user_id in the parameter...
db.query(User).delete() # ...but you never USE it!
db.commit()
return {"status": "deleted"}Think about it...
Why does db.query(User).filter(User.id == 5).first() return None instead of raising an exception when the user doesn't exist?
Hint: Think about what 'first' means when there are zero items.
Key Points
Repository Pattern
Isolate DB logic in crud.py — keep endpoints in routers
model_dump()
Convert Pydantic schemas to dicts for ORM object creation
Partial Updates
Use exclude_unset=True to only update fields the client sent
404 Checks
Always check if the object exists before update or delete