SQLAlchemy là một thư viện ORM mạnh mẽ cho phép thao tác cơ sở dữ liệu theo hướng đối tượng trong Python. Dưới đây là cách triển khai đầy đủ từ thiết lập đến các thao tác nâng cao.

Mục lục

1. Cài đặt môi trường
2. Các thành phần cốt lõi
3. Thiết lập kết nối
4. Xây dựng mô hình dữ liệu
5. Tạo bảng tự động
6. Thao tác CRUD cơ bản
7. Truy vấn linh hoạt
8. Xử lý quan hệ giữa bảng
9. Quản lý giao dịch
10. Thực hành tối ưu

Cài đặt

pip install sqlalchemy

# Với PostgreSQL
pip install psycopg2-binary

# Với MySQL
pip install mysql-connector-python

Thành phần cốt lõi

• Engine: Quản lý kết nối vật lý tới CSDL
• Session: Đơn vị làm việc với dữ liệu, hỗ trợ commit/rollback
• DeclarativeBase: Lớp cha để định nghĩa model
• Query Builder: Công cụ xây dựng câu truy vấn động

Thiết lập kết nối

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

db_engine = create_engine('sqlite:///app.db', echo=False)
SessionFactory = sessionmaker(bind=db_engine, expire_on_commit=False)
current_session = SessionFactory()

Xây dựng mô hình dữ liệu

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase

class Base(DeclarativeBase):
    pass

class Author(Base):
    __tablename__ = 'authors'
    
    uid = Column(Integer, primary_key=True)
    full_name = Column(String(80), nullable=False)
    contact = Column(String(120), unique=True)
    
    articles = relationship("Article", back_populates="writer")

class Article(Base):
    __tablename__ = 'articles'
    
    aid = Column(Integer, primary_key=True)
    headline = Column(String(150), nullable=False)
    body = Column(String)
    writer_id = Column(Integer, ForeignKey('authors.uid'))
    
    writer = relationship("Author", back_populates="articles")
    categories = relationship("Category", secondary="article_category", back_populates="entries")

class Category(Base):
    __tablename__ = 'categories'
    
    cid = Column(Integer, primary_key=True)
    label = Column(String(40), unique=True)
    
    entries = relationship("Article", secondary="article_category", back_populates="categories")

class ArticleCategory(Base):
    __tablename__ = 'article_category'
    article_id = Column(Integer, ForeignKey('articles.aid'), primary_key=True)
    category_id = Column(Integer, ForeignKey('categories.cid'), primary_key=True)

Tạo bảng tự động

Base.metadata.create_all(db_engine)

Thao tác CRUD

Thêm dữ liệu

new_author = Author(full_name="Nguyễn Văn A", contact="a@email.com")
current_session.add(new_author)
current_session.commit()

# Thêm nhiều bản ghi
current_session.add_all([
    Author(full_name="Lê Thị B", contact="b@email.com"),
    Author(full_name="Trần Văn C", contact="c@email.com")
])
current_session.commit()

Đọc dữ liệu

all_authors = current_session.query(Author).all()
first_author = current_session.query(Author).first()
author_by_id = current_session.query(Author).get(1)

Cập nhật dữ liệu

target = current_session.query(Author).get(1)
target.full_name = "Nguyễn Văn A (đã cập nhật)"
current_session.commit()

# Cập nhật hàng loạt
current_session.query(Author).filter(Author.full_name.like("Nguyễn%")).update(
    {"full_name": "Họ Nguyễn"}, synchronize_session=False
)
current_session.commit()

Xóa dữ liệu

to_remove = current_session.query(Author).get(2)
current_session.delete(to_remove)
current_session.commit()

# Xóa theo điều kiện
current_session.query(Author).filter(Author.full_name == "Lê Thị B").delete(synchronize_session=False)
current_session.commit()

Truy vấn nâng cao

Lọc và sắp xếp

from sqlalchemy import or_

# Tìm theo tên
results = current_session.query(Author).filter(Author.full_name == "Nguyễn Văn A").all()

# Tìm theo mẫu
matches = current_session.query(Author).filter(Author.full_name.like("%Văn%")).all()

# Sắp xếp giảm dần
ordered = current_session.query(Author).order_by(Author.full_name.desc()).all()

Truy vấn tổng hợp

from sqlalchemy import func

total = current_session.query(Author).count()
average_id = current_session.query(func.avg(Author.uid)).scalar()

# Nhóm và đếm
stats = current_session.query(
    Author.full_name,
    func.count(Article.aid)
).join(Article).group_by(Author.full_name).all()

Liên kết bảng

# INNER JOIN
joined_data = current_session.query(Author, Article).join(Article).filter(
    Article.headline.contains("Python")
).all()

# LEFT JOIN
left_joined = current_session.query(Author, Article).outerjoin(Article).all()

Xử lý quan hệ

author_instance = Author(full_name="Phạm Minh D", contact="d@email.com")
article_instance = Article(headline="Giới thiệu SQLAlchemy", body="Nội dung...", writer=author_instance)

tech_tag = Category(label="Công nghệ")
db_tag = Category(label="Cơ sở dữ liệu")

article_instance.categories.append(tech_tag)
article_instance.categories.append(db_tag)

current_session.add(article_instance)
current_session.commit()

# Truy cập qua quan hệ
print(f"Bài viết '{article_instance.headline}' do {article_instance.writer.full_name} viết")
print("Thuộc các danh mục:")
for cat in article_instance.categories:
    print(f" - {cat.label}")

Quản lý giao dịch

try:
    new_entry = Author(full_name="Người dùng thử", contact="test@email.com")
    current_session.add(new_entry)
    current_session.commit()
except Exception as error:
    current_session.rollback()
    print(f"Lỗi: {error}")

# Sử dụng context manager
from contextlib import contextmanager

@contextmanager
def db_transaction():
    sess = SessionFactory()
    try:
        yield sess
        sess.commit()
    except:
        sess.rollback()
        raise
    finally:
        sess.close()

with db_transaction() as tx:
    tx.add(Author(full_name="Dùng Context", contact="ctx@email.com"))

Thực hành tối ưu

• Luôn đóng session sau khi dùng xong
• Sử dụng eager loading để tránh N+1 query
• Cấu hình connection pool phù hợp với tải hệ thống
• Kiểm tra ràng buộc dữ liệu trước khi commit
• Log lỗi chi tiết trong quá trình rollback