CRUD là bộ bốn thao tác tạo – đọc – sửa – xóa mà gần như mọi API đều cần. Cách chuẩn trong FastAPI là gom logic database vào các hàm CRUD dùng SQLAlchemy 2.0 — select(), db.get(), db.add(), db.delete() — lấy session qua dependency yield tên get_db, và khai báo response_model cùng status_code rõ ràng ở từng endpoint. Mẫu này dùng được cho mọi bảng, dễ test bằng TestClient, và tránh được session rò rỉ lẫn dữ liệu nhạy cảm lọt ra response.

Vấn đề

Nhiều tutorial vẫn viết CRUD bằng cách mở session trực tiếp trong endpoint, hoặc dùng chung một session global cho mọi request. Session global không an toàn khi server chạy đa luồng: hai request có thể chen nhau commit, dữ liệu chưa commit của request này lọt sang request khác, và session không bao giờ được đóng nên connection pool dần cạn. Kiểu cũ Model.query cũng đã bị bỏ trong SQLAlchemy 2.0 — code cũ sẽ lỗi khi nâng cấp.

# 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

Giải pháp

Chia code thành ba lớp rõ ràng. Lớp model định nghĩa bảng, lớp schema định nghĩa dữ liệu vào/ra, lớp CRUD chứa toàn bộ câu lệnh SQL. Endpoint chỉ gọi hàm CRUD và trả về đúng 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)
# 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)
# 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

Session được cấp phát qua dependency yield — FastAPI tự đóng session sau khi request xong, dù endpoint có lỗi hay không:

# 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()

Endpoint chỉ còn việc gọi hàm CRUD, khai báo response rõ ràng và trả 404 khi không tìm thấy bản ghi:

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

Ba chi tiết đáng nhớ: select() thay cho Model.query đã bị bỏ; db.get() trả về None nên endpoint tự quyết định trả 404; db.refresh() sau commit() để lấy lại id vừa sinh. Endpoint DELETE trả 204 với body rỗng — dùng Response(status_code=204) vì trả None kèm response_model sẽ lỗi.

Áp dụng vào dự án của bạn

Áp dụng mẫu này cho mọi resource trong API: mỗi bảng có một module CRUD riêng, endpoint chỉ gọi hàm CRUD và khai báo response_model. Khi cần phân trang, thêm skip/limit vào hàm get_items thay vì sửa endpoint; khi cần test, override dependency get_db bằng session SQLite in-memory — xem bài TestClient đã đăng.

Bài viết liên quan

Câu hỏi thường gặp

Vì sao không dùng một session global cho mọi request?

Session của SQLAlchemy không an toàn với đa luồng: hai request dùng chung một session có thể chen nhau commit, đọc nhầm dữ liệu chưa commit của nhau, và session không được đóng sẽ giữ connection tới khi pool cạn. Dependency yield cấp một session riêng cho mỗi request và đóng nó trong finally nên không gặp ba lỗi này.

select() khác gì Model.query?

Model.query là API kiểu cũ, đã bị loại bỏ hoàn toàn trong SQLAlchemy 2.0. select() là API chuẩn mới: viết được câu lệnh phức tạp như join, filter, order ngay trong một biểu thức, và kết quả lấy qua .scalars().all() hoặc .scalar_one_or_none() rõ ràng hơn.

Khi nào cần db.refresh() sau commit?

Sau commit(), SQLAlchemy hết hạn các thuộc tính của object. Gọi db.refresh() ngay sau commit để nạp lại id vừa sinh và các giá trị mặc định từ database trước khi trả object về cho client.

Có cần dùng AsyncSession thay Session thường không?

Session đồng bộ đủ cho hầu hết API. Chỉ chuyển sang AsyncSession khi bạn dùng async driver như asyncpg hoặc aiosqlite, và tuyệt đối không trộn Session thường với endpoint async def vì nó sẽ chặn event loop.

💬 Bạn đã thử cách này trong dự án của mình chưa? Nếu gặp lỗi hoặc có cách làm gọn hơn, để lại bình luận bên dưới — mình đọc hết và sẽ bổ sung những ý hay vào bài.

Nguồn tham khảo