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

1. Cài đặt SQLAlchemy

Để bắt đầu, bạn cần cài đặt thư viện SQLAlchemy. Mở terminal hoặc command prompt và chạy lệnh sau:

pip install sqlalchemy

Nếu bạn muốn kết nối với các hệ quản trị cơ sở dữ liệu cụ thể, bạn cũng cần cài đặt driver tương ứng:

• PostgreSQL:

pip install psycopg2-binary

• MySQL:

pip install mysqlclient

• SQLite: (Đã có sẵn trong thư viện chuẩn của Python, không cần cài đặt thêm)

2. Các Khái niệm Chính

Hiểu rõ các khái niệm sau sẽ giúp bạn làm việc hiệu quả với SQLAlchemy:

• Engine: Bộ máy kết nối, là điểm truy cập chính đến cơ sở dữ liệu, quản lý các kết nối và giao tiếp.
• Session: Phiên làm việc, chịu trách nhiệm quản lý tất cả các đối tượng (model) và thao tác truy vấn, đảm bảo tính nhất quán của dữ liệu.
• Declarative Base/Model: Lớp cơ sở để định nghĩa các mô hình dữ liệu, mỗi lớp con sẽ ánh xạ tới một bảng trong cơ sở dữ liệu.
• Query: Đối tượng truy vấn, được sử dụng để xây dựng và thực thi các câu lệnh SQL dưới dạng đối tượng Python.

3. Thiết lập Kết nối Cơ sở dữ liệu

Bạn cần tạo một Engine để thiết lập kết nối và một Session để tương tác:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Tạo engine kết nối cơ sở dữ liệu
# Ví dụ với SQLite (sẽ tạo một file du_lieu.db)
db_engine = create_engine('sqlite:///du_lieu.db', echo=True)

# Ví dụ với PostgreSQL
# db_engine = create_engine('postgresql://user_name:password@localhost:5432/ten_database')

# Ví dụ với MySQL
# db_engine = create_engine('mysql+mysqlconnector://user_name:password@localhost:3306/ten_database')

# Tạo một lớp Session, được cấu hình để không tự động commit hoặc flush
PhiênCSDL = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

# Tạo một phiên làm việc để sử dụng
# Lưu ý: Mỗi yêu cầu/luồng nên có một phiên riêng

4. Định nghĩa Mô hình Dữ liệu

Mô hình dữ liệu (Model) là các lớp Python ánh xạ tới các bảng trong cơ sở dữ liệu. Chúng được định nghĩa bằng cách kế thừa từ một lớp Base:

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

# Khởi tạo Base cho các mô hình khai báo
Base = declarative_base()

class NguoiDung(Base):
    __tablename__ = 'nguoi_dung' # Tên bảng trong CSDL
    
    id = Column(Integer, primary_key=True, index=True)
    ten_nguoi_dung = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # Định nghĩa mối quan hệ "một-nhiều" với BaiViet
    bai_viet = relationship("BaiViet", back_populates="tac_gia")

class BaiViet(Base):
    __tablename__ = 'bai_viet'
    
    id = Column(Integer, primary_key=True, index=True)
    tieu_de = Column(String(100), nullable=False)
    noi_dung = Column(String(500))
    ma_tac_gia = Column(Integer, ForeignKey('nguoi_dung.id'))
    
    # Định nghĩa mối quan hệ "nhiều-một" với NguoiDung
    tac_gia = relationship("NguoiDung", back_populates="bai_viet")
    
    # Định nghĩa mối quan hệ "nhiều-nhiều" với The (thông qua bảng liên kết)
    the_tags = relationship("The", secondary="bai_viet_the", back_populates="cac_bai_viet")

class The(Base):
    __tablename__ = 'the'
    
    id = Column(Integer, primary_key=True, index=True)
    ten_the = Column(String(30), unique=True, nullable=False)
    
    cac_bai_viet = relationship("BaiViet", secondary="bai_viet_the", back_populates="the_tags")

# Bảng liên kết cho mối quan hệ "nhiều-nhiều" giữa BaiViet và The
class BaiVietThe(Base):
    __tablename__ = 'bai_viet_the'
    
    ma_bai_viet = Column(Integer, ForeignKey('bai_viet.id'), primary_key=True)
    ma_the = Column(Integer, ForeignKey('the.id'), primary_key=True)

5. Tạo Bảng trong Cơ sở dữ liệu

Sau khi định nghĩa các mô hình, bạn có thể tạo tất cả các bảng tương ứng trong cơ sở dữ liệu:

# Tạo tất cả các bảng đã định nghĩa
Base.metadata.create_all(bind=db_engine)

# Để xóa tất cả các bảng (cần cẩn trọng khi sử dụng trong môi trường production)
# Base.metadata.drop_all(bind=db_engine)

6. Thao tác CRUD Cơ bản (Tạo, Đọc, Cập nhật, Xóa)

Sử dụng một phiên làm việc để thực hiện các thao tác dữ liệu.

Tạo dữ liệu mới

# Tạo một phiên làm việc
db_session = PhiênCSDL()

try:
    # Thêm một người dùng mới
    nguoi_dung_moi = NguoiDung(ten_nguoi_dung="Nguyễn Văn A", email="nguyenvana@example.com")
    db_session.add(nguoi_dung_moi)
    db_session.commit() # Lưu thay đổi vào CSDL
    print(f"Người dùng '{nguoi_dung_moi.ten_nguoi_dung}' đã được thêm với ID: {nguoi_dung_moi.id}")

    # Thêm nhiều người dùng cùng lúc
    db_session.add_all([
        NguoiDung(ten_nguoi_dung="Trần Thị B", email="tranb@example.com"),
        NguoiDung(ten_nguoi_dung="Lê Văn C", email="lec@example.com")
    ])
    db_session.commit()
    print("Thêm nhiều người dùng thành công.")

except Exception as e:
    db_session.rollback() # Hoàn tác nếu có lỗi
    print(f"Lỗi khi thêm dữ liệu: {e}")
finally:
    db_session.close() # Đóng phiên

Đọc dữ liệu

db_session = PhiênCSDL()
try:
    # Lấy tất cả người dùng
    tat_ca_nguoi_dung = db_session.query(NguoiDung).all()
    print("\n--- Tất cả người dùng ---")
    for nd in tat_ca_nguoi_dung:
        print(f"ID: {nd.id}, Tên: {nd.ten_nguoi_dung}, Email: {nd.email}")

    # Lấy người dùng đầu tiên
    nguoi_dung_dau_tien = db_session.query(NguoiDung).first()
    if nguoi_dung_dau_tien:
        print(f"\n--- Người dùng đầu tiên ---")
        print(f"ID: {nguoi_dung_dau_tien.id}, Tên: {nguoi_dung_dau_tien.ten_nguoi_dung}")

    # Lấy người dùng theo ID
    nguoi_dung_theo_id = db_session.query(NguoiDung).filter_by(id=1).first()
    if nguoi_dung_theo_id:
        print(f"\n--- Người dùng có ID 1 ---")
        print(f"ID: {nguoi_dung_theo_id.id}, Tên: {nguoi_dung_theo_id.ten_nguoi_dung}")

except Exception as e:
    print(f"Lỗi khi đọc dữ liệu: {e}")
finally:
    db_session.close()

Cập nhật dữ liệu

db_session = PhiênCSDL()
try:
    # Tìm và cập nhật một người dùng
    nguoi_dung_can_cap_nhat = db_session.query(NguoiDung).filter_by(email="nguyenvana@example.com").first()
    if nguoi_dung_can_cap_nhat:
        nguoi_dung_can_cap_nhat.ten_nguoi_dung = "Nguyễn Văn An"
        db_session.commit()
        print(f"\nCập nhật tên người dùng ID {nguoi_dung_can_cap_nhat.id} thành '{nguoi_dung_can_cap_nhat.ten_nguoi_dung}'")

    # Cập nhật hàng loạt (không tải đối tượng vào phiên)
    db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung.like("Lê%")).update(
        {"ten_nguoi_dung": "Lê Hoàn"}, synchronize_session=False
    )
    db_session.commit()
    print("\nCập nhật hàng loạt người dùng tên bắt đầu bằng 'Lê' thành 'Lê Hoàn'.")

except Exception as e:
    db_session.rollback()
    print(f"Lỗi khi cập nhật dữ liệu: {e}")
finally:
    db_session.close()

Xóa dữ liệu

db_session = PhiênCSDL()
try:
    # Tìm và xóa một người dùng
    nguoi_dung_can_xoa = db_session.query(NguoiDung).filter_by(email="tranb@example.com").first()
    if nguoi_dung_can_xoa:
        db_session.delete(nguoi_dung_can_xoa)
        db_session.commit()
        print(f"\nĐã xóa người dùng có ID: {nguoi_dung_can_xoa.id}")

    # Xóa hàng loạt
    db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung == "Lê Hoàn").delete(synchronize_session=False)
    db_session.commit()
    print("\nĐã xóa tất cả người dùng có tên 'Lê Hoàn'.")

except Exception as e:
    db_session.rollback()
    print(f"Lỗi khi xóa dữ liệu: {e}")
finally:
    db_session.close()

7. Các Loại Truy vấn

SQLAlchemy cung cấp một API truy vấn phong phú để lọc, sắp xếp, nhóm và kết nối dữ liệu.

Truy vấn cơ bản

db_session = PhiênCSDL()
try:
    # Lấy tất cả bản ghi
    ds_nguoi_dung = db_session.query(NguoiDung).all()

    # Lấy các trường cụ thể
    danh_sach_ten = db_session.query(NguoiDung.ten_nguoi_dung).all()
    print(f"\nDanh sách tên người dùng: {[ten for ten, in danh_sach_ten]}")

    # Sắp xếp kết quả
    nguoi_dung_sap_xep = db_session.query(NguoiDung).order_by(NguoiDung.ten_nguoi_dung.asc()).all()
    print("\nNgười dùng sắp xếp theo tên (tăng dần):")
    for nd in nguoi_dung_sap_xep:
        print(f"- {nd.ten_nguoi_dung}")

    # Giới hạn số lượng kết quả
    top_5_nguoi_dung = db_session.query(NguoiDung).limit(5).all()

    # Phân trang với offset và limit
    trang_thu_hai = db_session.query(NguoiDung).offset(5).limit(5).all()

except Exception as e:
    print(f"Lỗi khi truy vấn cơ bản: {e}")
finally:
    db_session.close()

Truy vấn có điều kiện (Filter)

from sqlalchemy import or_

db_session = PhiênCSDL()
try:
    # Lọc bằng giá trị cụ thể
    nd_tim_thay = db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung == "Nguyễn Văn An").first()
    if nd_tim_thay:
        print(f"\nTìm thấy người dùng: {nd_tim_thay.ten_nguoi_dung}")

    # Lọc theo mẫu (LIKE)
    nguoi_dung_like = db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung.like("Nguyễn%")).all()
    print(f"\nNgười dùng có tên bắt đầu bằng 'Nguyễn': {[nd.ten_nguoi_dung for nd in nguoi_dung_like]}")

    # Lọc với điều kiện IN
    nguoi_dung_in = db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung.in_(["Nguyễn Văn An", "Lê Hoàn"])).all()
    print(f"\nNgười dùng có tên trong danh sách: {[nd.ten_nguoi_dung for nd in nguoi_dung_in]}")

    # Lọc với nhiều điều kiện AND
    nguoi_dung_and = db_session.query(NguoiDung).filter(
        NguoiDung.ten_nguoi_dung == "Nguyễn Văn An",
        NguoiDung.email.like("%@example.com")
    ).first()

    # Lọc với điều kiện OR
    nguoi_dung_or = db_session.query(NguoiDung).filter(
        or_(NguoiDung.ten_nguoi_dung == "Nguyễn Văn An", NguoiDung.email == "notfound@example.com")
    ).all()
    print(f"\nNgười dùng thỏa mãn điều kiện OR: {[nd.ten_nguoi_dung for nd in nguoi_dung_or]}")

    # Lọc với điều kiện KHÔNG BẰNG
    nguoi_dung_khac = db_session.query(NguoiDung).filter(NguoiDung.ten_nguoi_dung != "Nguyễn Văn An").all()
    print(f"\nNgười dùng KHÔNG phải 'Nguyễn Văn An': {[nd.ten_nguoi_dung for nd in nguoi_dung_khac]}")

except Exception as e:
    print(f"Lỗi khi truy vấn có điều kiện: {e}")
finally:
    db_session.close()

Truy vấn tổng hợp (Aggregate)

from sqlalchemy import func

db_session = PhiênCSDL()
try:
    # Đếm tổng số người dùng
    tong_nguoi_dung = db_session.query(NguoiDung).count()
    print(f"\nTổng số người dùng: {tong_nguoi_dung}")

    # Thêm một bài viết mẫu để test group by
    tac_gia_a = db_session.query(NguoiDung).filter_by(ten_nguoi_dung="Nguyễn Văn An").first()
    if not tac_gia_a:
        tac_gia_a = NguoiDung(ten_nguoi_dung="Nguyễn Văn An", email="an@example.com")
        db_session.add(tac_gia_a)
        db_session.commit() # Commit để có ID
    
    bai_viet_1 = BaiViet(tieu_de="Bài viết của An 1", noi_dung="Nội dung...", tac_gia=tac_gia_a)
    bai_viet_2 = BaiViet(tieu_de="Bài viết của An 2", noi_dung="Nội dung khác...", tac_gia=tac_gia_a)
    db_session.add_all([bai_viet_1, bai_viet_2])
    db_session.commit()


    # Đếm số bài viết của mỗi người dùng (GROUP BY)
    so_bai_viet_moi_nguoi = db_session.query(
        NguoiDung.ten_nguoi_dung, 
        func.count(BaiViet.id)
    ).join(BaiViet).group_by(NguoiDung.ten_nguoi_dung).all()
    print("\nSố bài viết của mỗi người dùng:")
    for ten, so_luong in so_bai_viet_moi_nguoi:
        print(f"- {ten}: {so_luong} bài viết")

    # Tính giá trị trung bình của ID (ví dụ)
    id_trung_binh = db_session.query(func.avg(NguoiDung.id)).scalar()
    print(f"\nID trung bình của người dùng: {id_trung_binh:.2f}")

except Exception as e:
    db_session.rollback()
    print(f"Lỗi khi truy vấn tổng hợp: {e}")
finally:
    db_session.close()

Truy vấn kết nối (Join)

db_session = PhiênCSDL()
try:
    # Inner Join: Lấy người dùng và các bài viết của họ
    ket_qua_join = db_session.query(NguoiDung, BaiViet).join(BaiViet).filter(BaiViet.tieu_de.like("%An%")).all()
    print("\nNgười dùng và bài viết liên quan (Inner Join):")
    for nguoi_dung, bai_viet in ket_qua_join:
        print(f"- {nguoi_dung.ten_nguoi_dung} viết: '{bai_viet.tieu_de}'")

    # Left Outer Join: Lấy tất cả người dùng, kể cả những người không có bài viết
    ket_qua_outer_join = db_session.query(NguoiDung, BaiViet).outerjoin(BaiViet).all()
    print("\nNgười dùng và bài viết (Outer Join):")
    for nguoi_dung, bai_viet in ket_qua_outer_join:
        tieu_de_bv = bai_viet.tieu_de if bai_viet else "Không có bài viết"
        print(f"- {nguoi_dung.ten_nguoi_dung} - Bài viết: '{tieu_de_bv}'")

    # Join với điều kiện tùy chỉnh
    ket_qua_custom_join = db_session.query(NguoiDung, BaiViet).join(BaiViet, NguoiDung.id == BaiViet.ma_tac_gia).all()

except Exception as e:
    print(f"Lỗi khi truy vấn kết nối: {e}")
finally:
    db_session.close()

8. Thao tác với Mối quan hệ

SQLAlchemy cho phép bạn dễ dàng làm việc với các mối quan hệ đã định nghĩa trong mô hình.

db_session = PhiênCSDL()
try:
    # Tạo đối tượng có mối quan hệ
    tac_gia_moi = NguoiDung(ten_nguoi_dung="Phạm Thị X", email="phamx@example.com")
    bai_viet_x = BaiViet(tieu_de="Kinh nghiệm du lịch", noi_dung="Một số mẹo hay...", tac_gia=tac_gia_moi)
    db_session.add(bai_viet_x)
    db_session.commit()
    print(f"\nBài viết '{bai_viet_x.tieu_de}' của tác giả '{bai_viet_x.tac_gia.ten_nguoi_dung}' đã được tạo.")

    # Truy cập thông qua mối quan hệ
    nguoi_dung_x = db_session.query(NguoiDung).filter_by(ten_nguoi_dung="Phạm Thị X").first()
    if nguoi_dung_x:
        print(f"\nCác bài viết của {nguoi_dung_x.ten_nguoi_dung}:")
        for bv in nguoi_dung_x.bai_viet:
            print(f"  - {bv.tieu_de}")

    # Thao tác với mối quan hệ nhiều-nhiều
    tag_python = The(ten_the="Python")
    tag_orm = The(ten_the="ORM")
    
    bai_viet_x.the_tags.append(tag_python)
    bai_viet_x.the_tags.append(tag_orm)
    db_session.commit()

    print(f"\nCác thẻ của bài viết '{bai_viet_x.tieu_de}':")
    for tag in bai_viet_x.the_tags:
        print(f"  - {tag.ten_the}")

except Exception as e:
    db_session.rollback()
    print(f"Lỗi khi thao tác mối quan hệ: {e}")
finally:
    db_session.close()

9. Quản lý Giao dịch (Transactions)

Giao dịch đảm bảo rằng một loạt các thao tác được thực hiện cùng nhau một cách nguyên tử (hoặc tất cả thành công, hoặc tất cả thất bại).

from sqlalchemy.exc import SQLAlchemyError
from contextlib import contextmanager

# Sử dụng context manager để quản lý phiên và giao dịch an toàn
@contextmanager
def lay_phien_db():
    session_instance = PhiênCSDL()
    try:
        yield session_instance
        session_instance.commit() # Tự động commit nếu không có lỗi
    except SQLAlchemyError:
        session_instance.rollback() # Tự động rollback nếu có lỗi
        raise
    finally:
        session_instance.close()

# Ví dụ sử dụng giao dịch
try:
    with lay_phien_db() as db:
        nd_test_1 = NguoiDung(ten_nguoi_dung="Người Dùng Test 1", email="test1@example.com")
        db.add(nd_test_1)

        nd_test_2 = NguoiDung(ten_nguoi_dung="Người Dùng Test 2", email="test2@example.com")
        db.add(nd_test_2)
        
        # Giả lập một lỗi để kiểm tra rollback
        # raise ValueError("Lỗi giả định để rollback") 

    print("\nGiao dịch thành công: Thêm 2 người dùng.")
except ValueError as ve:
    print(f"Lỗi trong giao dịch: {ve}. Các thay đổi đã được rollback.")
except SQLAlchemyError as sqle:
    print(f"Lỗi cơ sở dữ liệu trong giao dịch: {sqle}. Các thay đổi đã được rollback.")

# Ví dụ về giao dịch lồng nhau (nested transaction / savepoint)
db_session = PhiênCSDL()
try:
    with db_session.begin_nested(): # Tạo một savepoint
        nd_savepoint = NguoiDung(ten_nguoi_dung="Người Dùng Savepoint", email="savepoint@example.com")
        db_session.add(nd_savepoint)
        print(f"\nĐã thêm người dùng '{nd_savepoint.ten_nguoi_dung}' vào savepoint.")
        # db_session.rollback() # Có thể rollback savepoint tại đây

    db_session.commit() # Commit các thay đổi bên ngoài và cả savepoint nếu chưa rollback
    print("Giao dịch savepoint hoàn tất.")

except Exception as e:
    db_session.rollback()
    print(f"Lỗi với savepoint: {e}. Đã rollback.")
finally:
    db_session.close()

10. Thực hành Tốt nhất

• Quản lý phiên: Luôn tạo một phiên mới cho mỗi yêu cầu hoặc luồng xử lý riêng biệt và đảm bảo đóng phiên sau khi hoàn tất để giải phóng tài nguyên.

  # Sử dụng context manager như đã minh họa trong phần giao dịch là cách tốt nhất.
  with lay_phien_db() as db:
      # Các thao tác với db
      pass
  

• Xử lý ngoại lệ: Luôn bọc các thao tác CSDL trong khối try...except...finally và gọi session.rollback() khi có lỗi, sau đó session.close().
• Lazy Loading và Eager Loading: Hiểu rõ cách SQLAlchemy tải dữ liệu các mối quan hệ để tránh vấn đề N+1 query (lazy loading mặc định) bằng cách sử dụng joinedload hoặc selectinload (eager loading) khi cần thiết.
• Cấu hình Connection Pool: Tùy chỉnh kích thước connection pool và thời gian chờ để tối ưu hiệu suất và quản lý tài nguyên hiệu quả trong ứng dụng lớn.
• Xác thực dữ liệu: Thực hiện xác thực dữ liệu ở cấp độ mô hình hoặc ứng dụng để đảm bảo tính toàn vẹn của dữ liệu trước khi lưu vào cơ sở dữ liệu.