SQLAlchemy là một thư viện ORM (Object-Relational Mapping) mạnh mẽ và linh hoạt cho Python, giúp chuyển đổi giữa các đối tượng Python và cấu trúc bảng quan hệ một cách tự nhiên. Bài viết này trình bày cách triển khai thực tế với SQLAlchemy — từ khởi tạo kết nối, định nghĩa mô hình dữ liệu, đến thao tác CRUD nâng cao, truy vấn phức tạp và kiểm soát giao dịch an toàn.

1. Cài đặt và thiết lập môi trường

Cài đặt cơ bản:

pip install sqlalchemy

Tùy vào hệ quản trị cơ sở dữ liệu (DBMS), cài thêm driver tương ứng:

• PostgreSQL: pip install psycopg2-binary
• MySQL/MariaDB: pip install mysqlclient hoặc pip install PyMySQL
• SQLite: Đã tích hợp sẵn trong Python — không cần cài thêm.

2. Các thành phần cốt lõi

• Engine: Đối tượng điều khiển kết nối vật lý tới cơ sở dữ liệu.
• Session: Đại diện cho một "phiên làm việc" với DB — nơi thực hiện mọi thao tác đọc/ghi.
• Declarative Base: Lớp nền để khai báo các mô hình (model classes).
• Query: Cơ chế xây dựng câu lệnh SELECT thông qua API hướng đối tượng.

3. Thiết lập kết nối và phiên làm việc

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Kết nối tới SQLite (dùng cho phát triển nhanh)
db_url = "sqlite:///app_data.db"

# Với PostgreSQL hoặc MySQL, dùng định dạng:
# db_url = "postgresql+psycopg2://user:pass@localhost:5432/myapp"
# db_url = "mysql+pymysql://user:pass@localhost:3306/myapp"

engine = create_engine(db_url, echo=False, pool_pre_ping=True)

# Tạo factory để sinh ra các phiên độc lập
Session = sessionmaker(bind=engine)
db_session = Session()

4. Định nghĩa mô hình dữ liệu với quan hệ đa chiều

from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey, Table
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime

Base = declarative_base()

# Bảng trung gian cho quan hệ nhiều-nhiều
article_category = Table(
    "article_category",
    Base.metadata,
    Column("article_id", Integer, ForeignKey("articles.id"), primary_key=True),
    Column("category_id", Integer, ForeignKey("categories.id"), primary_key=True),
)

class Author(Base):
    __tablename__ = "authors"
    
    id = Column(Integer, primary_key=True)
    full_name = Column(String(120), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    articles = relationship("Article", back_populates="writer", cascade="all, delete-orphan")

class Category(Base):
    __tablename__ = "categories"
    
    id = Column(Integer, primary_key=True)
    label = Column(String(60), unique=True, index=True)
    
    articles = relationship(
        "Article",
        secondary=article_category,
        back_populates="categories"
    )

class Article(Base):
    __tablename__ = "articles"
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    body = Column(String(2000))
    published = Column(Boolean, default=False)
    written_at = Column(DateTime, default=datetime.utcnow)
    
    author_id = Column(Integer, ForeignKey("authors.id"))
    writer = relationship("Author", back_populates="articles")
    
    categories = relationship(
        "Category",
        secondary=article_category,
        back_populates="articles"
    )

5. Sinh lược đồ cơ sở dữ liệu

# Chạy một lần để tạo bảng
Base.metadata.create_all(engine)

# (Tùy chọn) Xóa toàn bộ bảng — chỉ dùng trong môi trường dev
# Base.metadata.drop_all(engine)

6. Thao tác CRUD cơ bản

Thêm bản ghi:

author = Author(full_name="Nguyễn Văn A", is_active=True)
db_session.add(author)
db_session.flush()  # Đảm bảo có ID trước khi gán cho article

article = Article(
    title="Giới thiệu SQLAlchemy",
    body="ORM mạnh mẽ cho Python...",
    writer=author,
    published=True
)
article.categories.append(Category(label="Lập trình"))
db_session.add(article)
db_session.commit()

Truy vấn đơn giản:

# Lấy tất cả bài viết chưa xuất bản
drafts = db_session.query(Article).filter_by(published=False).all()

# Lấy tác giả cùng số bài viết của họ
from sqlalchemy import func
stats = db_session.query(
    Author.full_name,
    func.count(Article.id).label("article_count")
).join(Article).group_by(Author.id).all()

Cập nhật và xóa:

# Cập nhật hàng loạt
db_session.query(Article).filter(Article.title.contains("Python")).update(
    {"published": True},
    synchronize_session=False
)

# Xóa theo điều kiện
db_session.query(Author).filter(Author.is_active == False).delete(
    synchronize_session=False
)
db_session.commit()

7. Truy vấn nâng cao

Join và subquery:

from sqlalchemy import exists

# Tìm tác giả có ít nhất một bài đã xuất bản
subq = db_session.query(Article.author_id).filter(Article.published == True).subquery()
active_authors = db_session.query(Author).filter(Author.id.in_(subq)).all()

# Join trái để lấy cả tác giả không có bài viết
result = db_session.query(Author, Article.title).outerjoin(Article).all()

Sử dụng hàm tổng hợp và phân nhóm:

from sqlalchemy import func

report = db_session.query(
    Category.label,
    func.count(Article.id).label("total_articles"),
    func.avg(func.length(Article.body)).label("avg_body_length")
).join(article_category).join(Article).group_by(Category.id).all()

8. Quản lý giao dịch một cách an toàn

from contextlib import contextmanager

@contextmanager
def transaction_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Sử dụng
with transaction_scope() as tx:
    a1 = Author(full_name="Tác giả mới")
    tx.add(a1)
    tx.flush()  # Đảm bảo ID được cấp
    
    art = Article(title="Bài viết thử nghiệm", writer=a1)
    tx.add(art)
    # Không gọi commit — sẽ tự động commit nếu không có ngoại lệ

9. Một số lưu ý thực tiễn

• Dùng session.expire_on_commit=False nếu cần truy cập thuộc tính sau khi commit.
• Tránh N+1 bằng cách dùng joinedload() hoặc selectinload() khi truy vấn quan hệ.
• Kích hoạt pool_pre_ping=True trong create_engine() để kiểm tra kết nối trước mỗi query.
• Với ứng dụng web, nên gắn Session vào request lifecycle — ví dụ: dùng dependency trong FastAPI hoặc middleware trong Flask.