SQLModel trong FastAPI: Xây Dựng API Type-Safe Không Cần Boilerplate Pydantic + SQLAlchemy

Quick Start: API SQLModel Đầu Tiên Trong 5 Phút

Sau sáu tháng chạy SQLModel trên production, nhận xét thật lòng của tôi là: đây là một trong những lựa chọn thư viện thực tế nhất tôi từng làm cho các dự án FastAPI. Không phải vì nó bắt mắt, mà vì nó loại bỏ phần nhàm chán nhất của việc phát triển API — duy trì song song cả Pydantic schema lẫn SQLAlchemy model đồng bộ với nhau.

Cài tất cả mọi thứ bạn cần chỉ với một lệnh:

pip install sqlmodel fastapi uvicorn

Một API user hoàn chỉnh, chạy được ngay:

from fastapi import FastAPI, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select
from typing import Optional

# Model duy nhất: vừa là ORM table vừa là Pydantic validation
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str

engine = create_engine("sqlite:///./app.db")
SQLModel.metadata.create_all(engine)

app = FastAPI()

@app.post("/users/", response_model=User)
def create_user(user: User):
    with Session(engine) as session:
        session.add(user)
        session.commit()
        session.refresh(user)
        return user

@app.get("/users/", response_model=list[User])
def list_users():
    with Session(engine) as session:
        return session.exec(select(User)).all()

Chạy với uvicorn main:app --reload và mở /docs. Bạn sẽ có một Swagger UI hoàn toàn được type hóa mà không cần cấu hình gì thêm. Flag table=True là thứ làm cho điều này trở nên khả thi. Nó đăng ký class với mapper registry của SQLAlchemy trong khi vẫn giữ nguyên toàn bộ hành vi validation của Pydantic.

Deep Dive: SQLModel Kết Nối Pydantic và SQLAlchemy Như Thế Nào

SQLModel được xây dựng trực tiếp trên cả hai thư viện. Khi bạn định nghĩa một class với table=True, metaclass sẽ đăng ký nó như một ORM table của SQLAlchemy. Không có flag đó, bạn chỉ có một Pydantic model thuần — cú pháp class giống nhau, nhưng hành vi runtime khác.

Sự phân biệt này quan trọng vì gần như lúc nào bạn cũng muốn các model riêng biệt cho API boundary và database schema. Đây là pattern tôi hiện dùng trong mọi dự án:

class UserBase(SQLModel):
    name: str
    email: str

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hashed_password: str  # Không bao giờ trả về trong API response

class UserCreate(UserBase):
    password: str  # Mật khẩu thô, chỉ nhận khi input

class UserRead(UserBase):
    id: int  # Đảm bảo không null trong response

Các endpoint của bạn sau đó sẽ dùng đúng model cho đúng ngữ cảnh:

@app.post("/users/", response_model=UserRead)
def create_user(user_data: UserCreate):
    hashed = hash_password(user_data.password)
    db_user = User(
        name=user_data.name,
        email=user_data.email,
        hashed_password=hashed
    )
    with Session(engine) as session:
        session.add(db_user)
        session.commit()
        session.refresh(db_user)
        return db_user

Ba class, một định nghĩa base — không có field nào bị trùng lặp. UserCreate xử lý input validation, UserRead định hình response khi serialize, và User map vào bảng database thực tế bao gồm các field nhạy cảm không bao giờ rời server. Tôi học được điều này theo cách cứng đầu nhất: một lần review bảo mật phát hiện ra hashed_password đang bị leak vào API response vì tôi dùng trực tiếp table model làm response type.

Nâng Cao: Relationships, Query và Migration

Định Nghĩa Quan Hệ Giữa Các Table

Relationships sử dụng API Relationship chuẩn của SQLAlchemy, được expose qua import của SQLModel. Đây là model Post liên kết với User:

from sqlmodel import Relationship
from typing import List

class Post(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    author_id: int = Field(foreign_key="user.id")
    author: Optional["User"] = Relationship(back_populates="posts")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    posts: List[Post] = Relationship(back_populates="author")

Query với join trông y hệt SQLAlchemy thuần — không cần học cú pháp mới:

@app.get("/users/{user_id}/posts")
def get_user_posts(user_id: int):
    with Session(engine) as session:
        user = session.get(User, user_id)
        if not user:
            raise HTTPException(status_code=404, detail="Không tìm thấy người dùng")
        return user.posts

Schema Migration với Alembic

SQLModel tích hợp hoàn toàn với Alembic. Bước quan trọng là trỏ Alembic vào SQLModel.metadata:

pip install alembic
alembic init migrations

Trong migrations/env.py, import tất cả table model để metadata được populate, rồi gán vào:

from sqlmodel import SQLModel
from app.models import User, Post  # Phải import tất cả table models

target_metadata = SQLModel.metadata

Tạo và áp dụng migration như bình thường:

alembic revision --autogenerate -m "thêm bảng posts"
alembic upgrade head

Cấu Hình Async Engine

Với các API cần throughput cao, SQLModel đi kèm với async session wrapper:

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

# Dùng asyncpg cho PostgreSQL trên production
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

@app.get("/users/")
async def list_users():
    async with AsyncSession(engine) as session:
        result = await session.exec(select(User))
        return result.all()

Cài aiosqlite cho SQLite hoặc asyncpg cho PostgreSQL làm async driver.

Mẹo Thực Chiến: Bài Học Từ Sáu Tháng Dùng Production

API bề mặt đủ nhỏ để học trong một ngày. Những chỗ gai góc chỉ xuất hiện khi bạn vượt qua các ví dụ tutorial — và một vài trong số chúng thực sự không hề rõ ràng.

Không Bao Giờ Dùng Table Model Trực Tiếp Làm Response Type

Dùng User ở khắp nơi rất hấp dẫn vì sự đơn giản. Hãy cưỡng lại điều đó. Table model thường chứa các field bạn không bao giờ muốn serialize — hash mật khẩu, flag audit nội bộ, timestamp soft-delete. Luôn route response qua một class UserRead riêng biệt.

Cẩn Thận DetachedInstanceError Với Lazy-Loaded Relationships

Lazy loading của SQLAlchemy âm thầm trì hoãn việc fetch relationship cho đến khi bạn truy cập nó. Một khi session đóng lại, truy cập vào relationship sẽ raise DetachedInstanceError. Cách fix là eager load trong phạm vi session:

from sqlalchemy.orm import selectinload

statement = select(User).options(selectinload(User.posts))
user = session.exec(statement).first()

Type Annotation Quyết Định Nullability Của Cột Database

SQLModel suy ra column constraint trực tiếp từ Python type hints. Field có kiểu str trở thành NOT NULL. Field có kiểu Optional[str] hoặc str | None trở thành nullable. Bỏ qua điều này và Alembic autogenerate của bạn sẽ âm thầm lệch khỏi schema thực tế — thường chỉ phát hiện ra vào lúc tệ nhất có thể.

Dùng In-Memory SQLite Để Test Nhanh, Cô Lập

import pytest
from sqlmodel import create_engine, Session, SQLModel
from fastapi.testclient import TestClient
from app.main import app, get_session

@pytest.fixture(name="session")
def session_fixture():
    engine = create_engine("sqlite:///:memory:")
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture(name="client")
def client_fixture(session: Session):
    def override_get_session():
        return session
    app.dependency_overrides[get_session] = override_get_session
    return TestClient(app)

Mỗi test có một schema sạch, độc lập mà không cần teardown.

Alembic Autogenerate Có Những Điểm Mù

Autogenerate không phát hiện được tất cả — custom constraint, server-side default và một số loại index cần script migration viết tay. Luôn review file được tạo ra trước khi chạy alembic upgrade head trên database production.

Sebastián Ramírez — tác giả FastAPI — cũng là người xây dựng SQLModel, điều này giải thích tại sao cả hai ăn khớp với nhau một cách tự nhiên đến vậy. Sau sáu tháng, team tôi chưa bao giờ cần dùng đến raw SQLAlchemy. Với bất kỳ dự án FastAPI mới nào, đây là điểm khởi đầu — không còn là thử nghiệm nữa.