Làm việc với cơ sở dữ liệu bằng SQLAlchemy 2.0

SQLAlchemy là thư viện làm việc với cơ sở dữ liệu phổ biến nhất trong hệ sinh thái Python ngoài Django. Nó cho phép bạn thao tác với dữ liệu qua đối tượng Python, đồng thời vẫn giữ khả năng viết SQL thuần khi cần.

Cài đặt

pip install sqlalchemy
pip install psycopg2-binary     # PostgreSQL
pip install pymysql             # MySQL / MariaDB

SQLite đã có sẵn trong Python nên không cần cài driver riêng — rất tiện khi thử nghiệm.

Định nghĩa mô hình

from datetime import datetime
from sqlalchemy import String, ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class TacGia(Base):
    __tablename__ = "tac_gia"

    id: Mapped[int] = mapped_column(primary_key=True)
    ten: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)

    bai_viet: Mapped[list["BaiViet"]] = relationship(back_populates="tac_gia")

class BaiViet(Base):
    __tablename__ = "bai_viet"

    id: Mapped[int] = mapped_column(primary_key=True)
    tieu_de: Mapped[str] = mapped_column(String(200), index=True)
    noi_dung: Mapped[str]
    ngay_dang: Mapped[datetime] = mapped_column(default=datetime.now)
    tac_gia_id: Mapped[int] = mapped_column(ForeignKey("tac_gia.id"))

    tac_gia: Mapped["TacGia"] = relationship(back_populates="bai_viet")

Cú pháp Mapped[...] là kiểu khai báo hiện đại của SQLAlchemy 2.0. Nó tận dụng gợi ý kiểu nên trình soạn thảo hiểu được cấu trúc dữ liệu và gợi ý code chính xác.

Tạo bảng và phiên làm việc

from sqlalchemy.orm import Session

engine = create_engine("postgresql+psycopg2://user:pass@localhost/blog")
Base.metadata.create_all(engine)

Trong thực tế, create_all chỉ dùng cho thử nghiệm. Dự án thật nên dùng Alembic để quản lý thay đổi cấu trúc bảng theo phiên bản.

Thêm dữ liệu

with Session(engine) as session:
    tac_gia = TacGia(ten="Nguyễn Văn A", email="a@example.com")
    bai = BaiViet(
        tieu_de="Bài viết đầu tiên",
        noi_dung="Nội dung...",
        tac_gia=tac_gia,
    )
    session.add(bai)
    session.commit()

Bạn chỉ cần add bài viết — SQLAlchemy tự nhận ra tác giả cũng là đối tượng mới và lưu cả hai theo đúng thứ tự.

Truy vấn

from sqlalchemy import select

with Session(engine) as session:
    stmt = select(BaiViet).where(BaiViet.tieu_de.like("%Python%"))
    for bai in session.scalars(stmt):
        print(bai.tieu_de)

    # Đếm
    from sqlalchemy import func
    tong = session.scalar(select(func.count()).select_from(BaiViet))

    # Sắp xếp và giới hạn
    moi_nhat = session.scalars(
        select(BaiViet).order_by(BaiViet.ngay_dang.desc()).limit(10)
    ).all()

Vấn đề N+1 và cách tránh

Đây là lỗi hiệu năng phổ biến nhất khi dùng ORM:

for bai in session.scalars(select(BaiViet)):
    print(bai.tac_gia.ten)     # mỗi vòng lặp phát sinh thêm 1 truy vấn

Với 100 bài viết, đoạn code trên chạy 101 câu lệnh SQL. Giải pháp là nạp sẵn quan hệ:

from sqlalchemy.orm import selectinload

stmt = select(BaiViet).options(selectinload(BaiViet.tac_gia))
for bai in session.scalars(stmt):
    print(bai.tac_gia.ten)     # chỉ 2 truy vấn cho toàn bộ

Muốn phát hiện sớm vấn đề này, hãy bật ghi log SQL trong môi trường phát triển:

engine = create_engine(url, echo=True)

Nhìn vào log, nếu bạn thấy cùng một câu SELECT lặp đi lặp lại với id khác nhau, đó chính là N+1.

Giao dịch

with Session(engine) as session:
    try:
        session.add(doi_tuong_1)
        session.add(doi_tuong_2)
        session.commit()
    except Exception:
        session.rollback()
        raise

Cách gọn hơn là dùng session.begin(), nó tự commit khi thoát khối lệnh bình thường và tự rollback khi có ngoại lệ:

with Session(engine) as session, session.begin():
    session.add(doi_tuong_1)
    session.add(doi_tuong_2)

Khi ORM không phù hợp

Với báo cáo phức tạp có nhiều phép nối và hàm tổng hợp, viết SQL thuần thường rõ ràng hơn:

from sqlalchemy import text

with engine.connect() as conn:
    ket_qua = conn.execute(text("""
        SELECT t.ten, COUNT(b.id) AS so_bai
        FROM tac_gia t
        LEFT JOIN bai_viet b ON b.tac_gia_id = t.id
        GROUP BY t.ten
        HAVING COUNT(b.id) > :nguong
        ORDER BY so_bai DESC
    """), {"nguong": 5})

    for dong in ket_qua:
        print(dong.ten, dong.so_bai)

Luôn truyền tham số qua dictionary như trên, đừng bao giờ ghép chuỗi SQL bằng f-string — đó là cửa ngõ của lỗ hổng SQL injection.

Kết nối trong ứng dụng web

Engine nên được tạo một lần khi ứng dụng khởi động, không tạo lại mỗi request. SQLAlchemy quản lý sẵn một pool kết nối:

engine = create_engine(
    url,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,      # kiểm tra kết nối còn sống trước khi dùng
    pool_recycle=3600,       # tái tạo kết nối sau 1 giờ
)

Tham số pool_pre_ping giải quyết vấn đề rất hay gặp: kết nối bị máy chủ database đóng do nhàn rỗi quá lâu, nhưng ứng dụng không biết và vẫn cố dùng lại.