Cài đặt và Thiết lập

Thiết lập môi trường bằng lệnh:

pip install sqlalchemy
# Driver cho PostgreSQL
pip install psycopg2-binary

Khái niệm Cốt lõi

• Engine: Thành phần quản lý kết nối cơ sở dữ liệu
• Session: Phiên làm việc để thực thi các thao tác
• Base: Lớp gốc cho các mô hình dữ liệu
• Relationship: Xác định liên kết giữa các bảng

Kết nối Cơ sở Dữ liệu

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Kết nối PostgreSQL
db_engine = create_engine(
    'postgresql://app_user:secure_pass@localhost:5432/app_db',
    pool_size=10,
    max_overflow=20
)

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

Mô hình Dữ liệu

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

Base = declarative_base()

class NguoiDung(Base):
    __tablename__ = 'nguoi_dung'
    
    ma = Column(Integer, primary_key=True)
    ten = Column(String(50), nullable=False)
    email = Column(String(100), unique=True)
    
    bai_viet = relationship("BaiViet", back_populates="tac_gia")

class BaiViet(Base):
    __tablename__ = 'bai_viet'
    
    ma = Column(Integer, primary_key=True)
    tieu_de = Column(String(100))
    noi_dung = Column(String)
    tac_gia_id = Column(Integer, ForeignKey('nguoi_dung.ma'))
    
    tac_gia = relationship("NguoiDung", back_populates="bai_viet")
    the = relationship("The", secondary="bai_viet_the", back_populates="cac_bai_viet")

class The(Base):
    __tablename__ = 'the'
    
    ma = Column(Integer, primary_key=True)
    ten = Column(String(30), unique=True)
    cac_bai_viet = relationship("BaiViet", secondary="bai_viet_the", back_populates="the")

class BaiVietThe(Base):
    __tablename__ = 'bai_viet_the'
    bai_viet_id = Column(Integer, ForeignKey('bai_viet.ma'), primary_key=True)
    the_id = Column(Integer, ForeignKey('the.ma'), primary_key=True)

Thao tác Dữ liệu Cơ bản

Chèn dữ liệu

# Tạo người dùng mới
nguoi_moi = NguoiDung(ten="Nguyễn Văn A", email="vana@example.com")
db_session.add(nguoi_moi)
db_session.commit()

# Chèn hàng loạt
db_session.bulk_save_objects([
    NguoiDung(ten="Trần Thị B", email="thib@example.com"),
    NguoiDung(ten="Lê Văn C", email="vanc@example.com")
])
db_session.commit()

Truy vấn Dữ liệu

# Lấy danh sách người dùng
tat_ca = db_session.query(NguoiDung).all()

# Tìm theo điều kiện
ket_qua = db_session.query(NguoiDung).filter(
    NguoiDung.ten.ilike("nguyễn%"), 
    NguoiDung.email.endswith("@example.com")
).limit(5).all()

# Truy vấn kết hợp
dulieu = db_session.query(NguoiDung.ten, BaiViet.tieu_de).join(
    BaiViet, NguoiDung.ma == BaiViet.tac_gia_id
).filter(BaiViet.tieu_de.contains("Python")).all()

Cập nhật và Xóa

# Cập nhật theo điều kiện
db_session.query(NguoiDung).filter(
    NguoiDung.email == "vana@example.com"
).update({"ten": "Nguyễn Văn An"}, synchronize_session=False)
db_session.commit()

# Xóa bản ghi
db_session.query(NguoiDung).filter(
    NguoiDung.ten == "Lê Văn C"
).delete(synchronize_session=False)
db_session.commit()

Quản lý Giao dịch

def them_bai_viet(ten_nguoi_dung, tieu_de, noi_dung):
    try:
        tac_gia = db_session.query(NguoiDung).filter_by(ten=ten_nguoi_dung).one()
        bai_moi = BaiViet(tieu_de=tieu_de, noi_dung=noi_dung, tac_gia=tac_gia)
        db_session.add(bai_moi)
        db_session.commit()
        return bai_moi
    except Exception as loi:
        db_session.rollback()
        raise ValueError(f"Lỗi giao dịch: {str(loi)}") from None

# Sử dụng savepoint
with db_session.begin_nested():
    nguoi_test = NguoiDung(ten="Giao dịch", email="gd@example.com")
    db_session.add(nguoi_test)

Tối ưu Hiệu năng

• Sử dụng selectinload để tránh N+1 query
• Cấu hình pool kết nối phù hợp với tải hệ thống
• Áp dụng yield_per() cho tập kết quả lớn
• Sử dụng expression language cho truy vấn phức tạp

from sqlalchemy.orm import selectinload

# Tải trước dữ liệu quan hệ
danh_sach = db_session.query(NguoiDung).options(
    selectinload(NguoiDung.bai_viet)
).all()