Search Topics

Search across all FastAPI topics

GitHub

CRUD Operations

Core Concept

Create, 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.

terminal
# 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 hours

Question

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
# 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.

crud.py
# 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 user

Read 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.

crud.py
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 user

Delete: 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.

crud.py
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 user

Watch 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.

Broken code
main.py
@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