DEV Community

Linh Fish Dev
Linh Fish Dev

Posted on Originally published at linhfish.dev

FastAPI: Write Standard CRUD with SQLAlchemy 2.0 — Session Dependency and select() in 5 Minutes

#python #fastapi #sqlalchemy #tutorial

CRUD — the four create, read, update, and delete operations — is something almost every API needs. The standard way to do it in FastAPI is to put the database logic in CRUD functions built on SQLAlchemy 2.0 — select(), db.get(), db.add(), db.delete() — get the session from a yield dependency named get_db, and declare a clear response_model and status_code on each endpoint. This pattern works for any table, is easy to test with TestClient, and avoids both leaked sessions and sensitive data slipping into responses.

The Problem

Many tutorials still write CRUD by opening a session directly inside the endpoint, or by sharing one global session across all requests. A global session is unsafe when the server runs with multiple threads: two requests can interleave their commits, uncommitted data from one request can leak into another, and a session that is never closed slowly drains the connection pool. The old Model.query style has also been removed in SQLAlchemy 2.0 — old code will break when you upgrade.

# KHÔNG NÊN: session global dùng chung cho mọi request
SessionLocal = sessionmaker(bind=engine)
db = SessionLocal()  # một session duy nhất, không bao giờ đóng
@app.post("/items/")
def create_item(item: ItemCreate):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    return db_item
Enter fullscreen mode Exit fullscreen mode

The Solution

Split the code into three clear layers. The model layer defines the table, the schema layer defines input and output data, and the CRUD layer holds all the SQL statements. Endpoints only call CRUD functions and return the right status code.

# models.py
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
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(default=0.0)
Enter fullscreen mode Exit fullscreen mode 
# schemas.py
from pydantic import BaseModel, ConfigDict
class ItemCreate(BaseModel):
    name: str
    price: float = 0.0
class ItemRead(ItemCreate):
    id: int
    model_config = ConfigDict(from_attributes=True)
Enter fullscreen mode Exit fullscreen mode 
# crud.py
from sqlalchemy import select
from sqlalchemy.orm import Session
from models import Item
from schemas import ItemCreate
def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.execute(select(Item).offset(skip).limit(limit)).scalars().all()
def get_item(db: Session, item_id: int):
    return db.get(Item, item_id)  # tra None neu khong ton tai
def create_item(db: Session, item: ItemCreate):
    db_item = Item(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)  # nap lai id vua sinh
    return db_item
def update_item(db: Session, item_id: int, item: ItemCreate):
    db_item = db.get(Item, item_id)
    if db_item is None:
        return None
    db_item.name = item.name
    db_item.price = item.price
    db.commit()
    db.refresh(db_item)
    return db_item
def delete_item(db: Session, item_id: int) -> bool:
    db_item = db.get(Item, item_id)
    if db_item is None:
        return False
    db.delete(db_item)
    db.commit()
    return True
Enter fullscreen mode Exit fullscreen mode

The session is provided through a yield dependency — FastAPI closes it automatically after the request finishes, whether or not the endpoint raised an error:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
engine = create_engine("sqlite:///./app.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
Enter fullscreen mode Exit fullscreen mode

Endpoints are then left with just calling CRUD functions, declaring the response clearly, and returning 404 when a record is not found:

# main.py
from fastapi import Depends, FastAPI, HTTPException, Response
from sqlalchemy.orm import Session
import crud
from database import Base, engine, get_db
from schemas import ItemCreate, ItemRead
Base.metadata.create_all(bind=engine)
app = FastAPI()
@app.get("/items/", response_model=list[ItemRead])
def list_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return crud.get_items(db, skip=skip, limit=limit)
@app.get("/items/{item_id}", response_model=ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = crud.get_item(db, item_id)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Khong tim thay item")
    return db_item
@app.post("/items/", response_model=ItemRead, status_code=201)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    return crud.create_item(db, item)
@app.put("/items/{item_id}", response_model=ItemRead)
def update_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = crud.update_item(db, item_id, item)
    if db_item is None:
        raise HTTPException(status_code=404, detail="Khong tim thay item")
    return db_item
@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    if not crud.delete_item(db, item_id):
        raise HTTPException(status_code=404, detail="Khong tim thay item")
    return Response(status_code=204)
Enter fullscreen mode Exit fullscreen mode

Three details worth remembering: select() replaces the removed Model.query; db.get() returns None, so the endpoint itself decides to return 404; and db.refresh() after commit() fetches back the freshly generated id. A DELETE endpoint returns 204 with an empty body — use Response(status_code=204), because returning None together with a response_model raises an error.

Apply It to Your Project

Apply this pattern to every resource in your API: each table gets its own CRUD module, and endpoints only call CRUD functions and declare a response_model. When you need pagination, add skip/limit to the get_items function instead of editing the endpoint; when you need tests, override the get_db dependency with an in-memory SQLite session — see the TestClient post already published.

Frequently Asked Questions

Why not use one global session for every request?

A SQLAlchemy session is not thread-safe: two requests sharing one session can interleave their commits, read each other’s uncommitted data, and a session that is never closed holds a connection until the pool runs dry. A yield dependency provides a separate session per request and closes it in finally, so none of those three problems occur.

How is select() different from Model.query?

Model.query is the legacy API, removed entirely in SQLAlchemy 2.0. select() is the new standard API: it can express complex statements with joins, filters, and ordering in a single expression, and its results are fetched through the clearer .scalars().all() or .scalar_one_or_none().

When do I need db.refresh() after commit?

After commit(), SQLAlchemy expires the object’s attributes. Call db.refresh() right after the commit to reload the freshly generated id and any database defaults before returning the object to the client.

Do I need AsyncSession instead of the regular Session?

The synchronous Session is enough for most APIs. Switch to AsyncSession only when you use an async driver such as asyncpg or aiosqlite, and never mix a regular Session with an async def endpoint, because it will block the event loop.

References

Top comments (0)