Hướng dẫn Không-Thất-Bại: Bắt đầu với Database Migrations (FastAPI × SQLAlchemy × Alembic)

✅ Tóm tắt (Kết luận & Bức tranh tổng thể trước tiên)

• Bạn có thể làm gì với bài viết này
  Học cách áp dụng thay đổi schema (migrations) cho cơ sở dữ liệu trong ứng dụng FastAPI một cách an toàn, tái lập và thân thiện với nhóm. Định nghĩa models bằng SQLAlchemy và sử dụng Alembic cho theo dõi lịch sử, tự động tạo diff, và rollback — với các bước triển khai cụ thể, sẵn sàng cho production.
• Các bước chính
  1. Chuẩn bị cấu trúc dự án → 2) Định nghĩa ORM models bằng SQLAlchemy → 3) Khởi tạo Alembic → 4) Tự động tạo revision → 5) upgrade/downgrade → 6) Tránh bẫy (quy ước đặt tên, xung đột merge, quirks của SQLite)
• Cơ sở dữ liệu mục tiêu
  Bắt đầu phát triển với SQLite và sau đó chuyển sang PostgreSQL hoặc MySQL với ít ma sát nhất.
• Lợi ích
  • Tránh “Ối, mình làm hỏng DB vì ALTER thủ công…”
  • Giữ lại lịch sử thay đổi để nhóm của bạn có thể tái tạo cùng trạng thái bất cứ lúc nào
  • Xây dựng quy trình ổn định, dễ dàng tích hợp vào CI/CD

🎯 Ai sẽ được lợi (Ví dụ)

• Dev solo (sinh viên năm 3, dự án web đầu tiên)
  Xây dựng ứng dụng ToDo với SQLite. Muốn thêm an toàn cột “due date” hoặc “priority” sau này. Có thể copy-paste lệnh chạy và xem trực quan thay đổi (lịch sử).
• Nhóm nhỏ (3 dev làm hợp đồng)
  Spec thay đổi hàng tuần, thay đổi DB thường xuyên xung đột, đồng đội không thể tái lập DB local của nhau. Cần quy ước đặt tên, theo dõi lịch sử, rollback để vận hành không sự cố.
• Startup đang chuẩn bị scale
  Bắt đầu với SQLite, sau đó chuyển sang PostgreSQL khi user tăng. Học quy trình vững chắc dựa trên autogenerated diffs để scale mượt mà.

1. Chuẩn bị: Cấu trúc dự án tối thiểu & Cài đặt

1.1 Cấu trúc thư mục

fastapi-db/
├─ app/
│  ├─ main.py
│  ├─ db.py
│  ├─ models.py
│  └─ schemas.py
├─ alembic/           # Generated by Alembic (after init)
├─ alembic.ini        # Alembic config
└─ .env               # DATABASE_URL, etc. (optional)

1.2 Các package cần thiết

python3 -m venv .venv && source .venv/bin/activate
pip install fastapi uvicorn sqlalchemy alembic pydantic "psycopg[binary]"  # Cho PostgreSQL về sau
# Chỉ với SQLite thì không cần psycopg

Thuật ngữ: Migration = Hệ thống theo dõi thay đổi schema DB bằng lập trình (bảng, cột, ràng buộc, …) để bạn có thể upgrade (áp dụng) hoặc downgrade (hoàn tác) an toàn và lặp lại.

1.3 Chọn Database URL (SQLite ổn cho dev)

• Development (local): sqlite:///./app.db
• Production (ví dụ PostgreSQL): postgresql+psycopg://USER:PASSWORD@HOST:PORT/DBNAME

Đặt trong .env để dễ dàng chuyển đổi.

Điểm chính

• Thiết lập cấu trúc trước → cài dependency
• Chuyển DB URL qua environment variables, bắt đầu với SQLite

2. Định nghĩa SQLAlchemy ORM Models (kiểu 2.x cho tương lai)

2.1 db.py: Engine & Session (phiên bản sync)

Alembic hoạt động tốt nhất với sync engine (dù app của bạn là async).

# app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

2.2 models.py: Declarative Base & Quy ước đặt tên

Quy ước đặt tên giúp diff tự động ổn định — rất quan trọng.

# app/models.py
from datetime import datetime
from sqlalchemy import String, DateTime, func, MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

naming_convention = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s"
}

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=naming_convention)

class Todo(Base):
    __tablename__ = "todos"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False, index=True)
    description: Mapped[str | None] = mapped_column(String(1000))
    is_done: Mapped[bool] = mapped_column(default=False, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )

Mẹo: server_default=func.now() dùng default của DB — hoạt động trong SQLite và chuyển sang PostgreSQL tự nhiên.

2.3 Để Alembic quản lý việc tạo bảng ban đầu

Không gọi Base.metadata.create_all() trực tiếp. Luôn tạo bảng qua Alembic upgrade để lịch sử nhất quán.

Điểm chính

• Dùng ORM SQLAlchemy 2.x có type-hint để chắc chắn
• Quy ước đặt tên giảm “nhiễu” diff
• Ngay từ đầu, để Alembic kiểm soát schema

3. Khởi tạo Alembic & Liên kết Metadata của App

3.1 Khởi tạo

alembic init alembic

Generates alembic/ and alembic.ini.

3.2 Sửa alembic.ini (DB URL)

Để sqlalchemy.url comment; load từ env vars để tránh lỗi ở production.

3.3 alembic/env.py: Chỉ tới metadata của App

Truyền Base.metadata của app vào target_metadata để –autogenerate hoạt động.

# alembic/env.py (excerpt)
import os
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

from app.models import Base
target_metadata = Base.metadata

def get_url():
    url = os.getenv("DATABASE_URL")
    if url:
        return url
    return config.get_main_option("sqlalchemy.url")

def run_migrations_offline():
    context.configure(
        url=get_url(),
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
        compare_server_default=True,
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    connectable = engine_from_config(
        {"sqlalchemy.url": get_url()},
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
            compare_server_default=True,
        )
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

Điểm chính

• Đặt target_metadata = Base.metadata
• Bật compare_type / compare_server_default
• Ưu tiên dùng env var cho URL

4. Tạo & Áp dụng Migration đầu tiên (Autogenerate)

4.1 Tạo revision ban đầu

alembic revision --autogenerate -m "create todos table"

Kiểm tra file được sinh trong alembic/versions/ — đảm bảo DDL đúng ý bạn.

4.2 Áp dụng vào DB

alembic upgrade head

SQLite giờ có bảng todos. Từ đây, mọi thay đổi schema đều được lưu lại trong lịch sử.

4.3 Thử rollback

alembic downgrade -1
alembic upgrade head

Luôn test upgrade/downgrade ở local trước khi production.

Điểm chính

• –autogenerate không phải phép màu — hãy review script
• Test round-trip migration để an toàn