Đã đăng vào 44 phút trước 2 phút đọc

FastAPI: viết CRUD chuẩn với SQLAlchemy 2.0 — session dependency và select() trong 5 phút

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.

Vấn đề

Nhiều tutorial vẫn 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 chen nhau commit, dữ liệu chưa commit lọt sang nhau, và session không bao giờ đóng nên connection pool dần cạn. Kiểu cũ Model.query cũng đã bị bỏ trong SQLAlchemy 2.0.

Giải pháp

Chia code thành ba lớp rõ ràng: model định nghĩa bảng, schema định nghĩa dữ liệu vào/ra, 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).

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

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.