SQLAlchemy là một trong những thư viện ORM (Object-Relational Mapping) phổ biến và mạnh mẽ nhất trong hệ sinh thái Python. Bài viết này hướng dẫn cách sử dụng SQLAlchemy ORM để tương tác với cơ sở dữ liệu một cách hiệu quả.

Mục lục

1. Cài đặt SQLAlchemy
2. Các khái niệm cốt lõi
3. Kết nối cơ sở dữ liệu
4. Định nghĩa mô hình dữ liệu
5. Tạo bảng trong cơ sở dữ liệu
6. Các thao tác CRUD cơ bản
7. Truy vấn dữ liệu
8. Xử lý quan hệ giữa các bảng
9. Quản lý giao dịch (transaction)
10. Các thực hành tốt

Cài đặt

pip install sqlalchemy

Nếu làm việc với cơ sở dữ liệu cụ thể, cần cài thêm driver tương ứng:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (đã có sẵn trong thư viện chuẩn của Python)

Các khái niệm cốt lõi

• Engine: Động cơ kết nối đến cơ sở dữ liệu.
• Session: Phiên làm việc quản lý các thao tác ghi/đọc dữ liệu.
• Model: Lớp đại diện cho bảng trong cơ sở dữ liệu.
• Query: Đối tượng dùng để xây dựng và thực thi truy vấn.

Kết nối cơ sở dữ liệu

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Ví dụ với SQLite
db_engine = create_engine('sqlite:///app.db', echo=True)

# Ví dụ với PostgreSQL
# db_engine = create_engine('postgresql://user:pass@localhost:5432/appdb')

# Ví dụ với MySQL
# db_engine = create_engine('mysql+mysqlconnector://user:pass@localhost:3306/appdb')

SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)
db_session = SessionFactory()

Định nghĩa mô hình dữ liệu

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

BaseModel = declarative_base()

class Account(BaseModel):
    __tablename__ = 'accounts'
    
    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(60), nullable=False)
    contact_email = Column(String(120), unique=True, index=True)
    
    articles = relationship("Article", back_populates="owner")

class Article(BaseModel):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True, index=True)
    headline = Column(String(150), nullable=False)
    body = Column(String(1000))
    owner_id = Column(Integer, ForeignKey('accounts.id'))
    
    owner = relationship("Account", back_populates="articles")
    categories = relationship("Category", secondary="article_category_links", back_populates="articles")

class Category(BaseModel):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True, index=True)
    label = Column(String(40), unique=True, nullable=False)
    
    articles = relationship("Article", secondary="article_category_links", back_populates="categories")

class ArticleCategoryLink(BaseModel):
    __tablename__ = 'article_category_links'
    
    article_id = Column(Integer, ForeignKey('articles.id'), primary_key=True)
    category_id = Column(Integer, ForeignKey('categories.id'), primary_key=True)

Tạo bảng trong cơ sở dữ liệu

BaseModel.metadata.create_all(bind=db_engine)

Các thao tác CRUD cơ bản

Tạo dữ liệu

new_account = Account(full_name="Nguyễn Văn A", contact_email="vana@example.com")
db_session.add(new_account)
db_session.commit()

db_session.add_all([
    Account(full_name="Trần Thị B", contact_email="thib@example.com"),
    Account(full_name="Lê Văn C", contact_email="vanc@example.com")
])
db_session.commit()

Đọc dữ liệu

all_accounts = db_session.query(Account).all()
first_acc = db_session.query(Account).first()
acc_by_id = db_session.query(Account).get(1)

Cập nhật dữ liệu

target_acc = db_session.query(Account).get(1)
target_acc.full_name = "Nguyễn Văn A (mới)"
db_session.commit()

db_session.query(Account).filter(Account.full_name.like("Nguyễn%")).update(
    {"full_name": "Họ Nguyễn"}, synchronize_session=False
)
db_session.commit()

Xóa dữ liệu

acc_to_delete = db_session.query(Account).get(1)
db_session.delete(acc_to_delete)
db_session.commit()

db_session.query(Account).filter(Account.full_name == "Trần Thị B").delete(synchronize_session=False)
db_session.commit()

Truy vấn dữ liệu

Truy vấn cơ bản

names_only = db_session.query(Account.full_name).all()
sorted_accs = db_session.query(Account).order_by(Account.full_name.desc()).all()
limited_accs = db_session.query(Account).limit(5).offset(2).all()

Lọc dữ liệu

from sqlalchemy import or_

exact_match = db_session.query(Account).filter(Account.full_name == "Nguyễn Văn A").first()
wildcard_search = db_session.query(Account).filter(Account.full_name.like("Nguyễn%")).all()
in_filter = db_session.query(Account).filter(Account.full_name.in_(["Nguyễn Văn A", "Lê Văn C"])).all()
multi_cond = db_session.query(Account).filter(
    Account.full_name == "Nguyễn Văn A",
    Account.contact_email.endswith("@example.com")
).all()
or_cond = db_session.query(Account).filter(
    or_(Account.full_name == "Nguyễn Văn A", Account.full_name == "Lê Văn C")
).all()

Truy vấn tổng hợp

from sqlalchemy import func

total_accounts = db_session.query(Account).count()
posts_per_user = db_session.query(
    Account.full_name,
    func.count(Article.id)
).join(Article).group_by(Account.full_name).all()
avg_id_val = db_session.query(func.avg(Account.id)).scalar()

Liên kết bảng (JOIN)

inner_join_res = db_session.query(Account, Article).join(Article).filter(
    Article.headline.contains("Python")
).all()

left_join_res = db_session.query(Account, Article).outerjoin(Article).all()

explicit_join_res = db_session.query(Account, Article).join(
    Article, Account.id == Article.owner_id
).all()

Xử lý quan hệ giữa các bảng

author = Account(full_name="Phạm Minh D", contact_email="minhd@example.com")
blog_post = Article(headline="Giới thiệu SQLAlchemy", body="ORM rất tiện lợi!", owner=author)
db_session.add(blog_post)
db_session.commit()

print(f"Bài viết '{blog_post.headline}' được viết bởi {blog_post.owner.full_name}")
for art in author.articles:
    print(f"- {art.headline}")

tag_python = Category(label="Python")
tag_sqlalchemy = Category(label="SQLAlchemy")
blog_post.categories.extend([tag_python, tag_sqlalchemy])
db_session.commit()

for cat in blog_post.categories:
    print(f"- {cat.label}")

Quản lý giao dịch (Transaction)

try:
    temp_acc = Account(full_name="Tài khoản thử", contact_email="test@example.com")
    db_session.add(temp_acc)
    db_session.commit()
except Exception as err:
    db_session.rollback()
    print(f"Lỗi: {err}")

# Giao dịch lồng nhau
with db_session.begin_nested():
    nested_acc = Account(full_name="Giao dịch con", contact_email="nested@example.com")
    db_session.add(nested_acc)

# Điểm lưu (savepoint)
sp = db_session.begin_nested()
try:
    sp_acc = Account(full_name="Dùng savepoint", contact_email="sp@example.com")
    db_session.add(sp_acc)
    sp.commit()
except:
    sp.rollback()

Các thực hành tốt

1. Quản lý phiên làm việc: Tạo phiên mới cho mỗi yêu cầu và đóng sau khi xử lý xong.
2. Xử lý ngoại lệ: Luôn gọi rollback() khi có lỗi.
3. Tránh N+1 query: Sử dụng joinedload hoặc selectinload để tải trước dữ liệu liên quan.
4. Cấu hình connection pool: Điều chỉnh kích thước và thời gian chờ phù hợp với tải hệ thống.
5. Xác thực dữ liệu: Kiểm tra tính hợp lệ trước khi lưu vào cơ sở dữ liệu.

from contextlib import contextmanager

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

# Sử dụng
with get_database_session() as db:
    acc = Account(full_name="Dùng context", contact_email="ctx@example.com")
    db.add(acc)