SQLAlchemy là một bộ công cụ SQL mạnh mẽ và thư viện Object-Relational Mapping (ORM) dành cho ngôn ngữ lập trình Python. Nó cung cấp một mô hình lập trình linh hoạt, giúp tách biệt logic nghiệp vụ khỏi các chi tiết kỹ thuật của hệ quản trị cơ sở dữ liệu (DBMS).

1. Thiết lập môi trường

Để bắt đầu sử dụng SQLAlchemy, bạn cần cài đặt thư viện chính thông qua pip:

pip install sqlalchemy

Tùy thuộc vào loại cơ sở dữ liệu bạn sử dụng, hãy cài đặt thêm driver tương ứng:

• PostgreSQL: pip install psycopg2
• MySQL: pip install pymysql
• SQLite: Đã được tích hợp sẵn trong Python.

2. Khởi tạo Engine và Session

Engine là trung tâm điều phối các kết nối đến database, trong khi Session đóng vai trò như một đơn vị làm việc (unit of work) để quản lý các giao dịch.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

# Khởi tạo kết nối tới SQLite (file: shop.db)
DATABASE_URL = "sqlite:///shop.db"
engine = create_engine(DATABASE_URL, echo=False)

# Tạo factory cho các phiên làm việc
SessionFactory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
db_session = scoped_session(SessionFactory)

3. Định nghĩa cấu trúc Model

Sử dụng cơ chế declarative_base để ánh xạ các class Python với các bảng trong cơ sở dữ liệu.

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

Base = declarative_base()

# Bảng phụ cho mối quan hệ n-n giữa Product và Category
product_categories = Table(
    'product_categories',
    Base.metadata,
    Column('product_id', Integer, ForeignKey('products.id'), primary_key=True),
    Column('category_id', Integer, ForeignKey('categories.id'), primary_key=True)
)

class Customer(Base):
    __tablename__ = 'customers'
    
    id = Column(Integer, primary_key=True)
    fullname = Column(String(100), nullable=False)
    email = Column(String(120), unique=True)
    
    # Quan hệ 1-n: Một khách hàng có nhiều đơn hàng
    orders = relationship("Order", back_populates="buyer")

class Order(Base):
    __tablename__ = 'orders'
    
    id = Column(Integer, primary_key=True)
    order_code = Column(String(20), unique=True)
    customer_id = Column(Integer, ForeignKey('customers.id'))
    
    buyer = relationship("Customer", back_populates="orders")

class Category(Base):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(50), nullable=False)
    
    items = relationship("Product", secondary=product_categories, back_populates="tags")

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True)
    sku = Column(String(50), unique=True)
    price = Column(Float, default=0.0)
    
    tags = relationship("Category", secondary=product_categories, back_populates="items")

# Khởi tạo bảng trong database
Base.metadata.create_all(engine)

4. Thực thi các thao tác CRUD cơ bản

Thêm mới dữ liệu (Create)

# Thêm khách hàng mới
new_customer = Customer(fullname="Nguyễn Văn A", email="vana@example.com")
db_session.add(new_customer)
db_session.commit()

# Thêm nhiều bản ghi cùng lúc
db_session.add_all([
    Customer(fullname="Trần Thị B", email="thib@example.com"),
    Customer(fullname="Lê Văn C", email="vanc@example.com")
])
db_session.commit()

Truy vấn dữ liệu (Read)

# Lấy tất cả khách hàng
all_customers = db_session.query(Customer).all()

# Lọc dữ liệu theo điều kiện
customer_b = db_session.query(Customer).filter_by(fullname="Trần Thị B").first()

# Truy vấn với điều kiện phức tạp
from sqlalchemy import or_
results = db_session.query(Customer).filter(
    or_(Customer.email.like("%@gmail.com"), Customer.id < 10)
).all()

Cập nhật dữ liệu (Update)

# Tìm và cập nhật
target_user = db_session.query(Customer).get(1)
if target_user:
    target_user.fullname = "Nguyễn Văn Chỉnh Sửa"
    db_session.commit()

# Cập nhật hàng loạt không cần load object
db_session.query(Product).filter(Product.price < 100).update({"price": 105.0})
db_session.commit()

Xóa dữ liệu (Delete)

# Xóa một đối tượng cụ thể
user_to_delete = db_session.query(Customer).filter_by(id=3).first()
if user_to_delete:
    db_session.delete(user_to_delete)
    db_session.commit()

5. Truy vấn nâng cao và Aggregation

SQLAlchemy hỗ trợ các hàm tổng hợp và kỹ thuật Join mạnh mẽ để xử lý dữ liệu phức tạp.

from sqlalchemy import func

# Đếm số lượng khách hàng
total_count = db_session.query(func.count(Customer.id)).scalar()

# Join hai bảng để lấy thông tin đơn hàng cùng tên khách hàng
order_details = db_session.query(Order.order_code, Customer.fullname)\
    .join(Customer, Order.customer_id == Customer.id)\
    .all()

# Gom nhóm và tính giá trị trung bình
avg_price_query = db_session.query(
    func.avg(Product.price).label('average')
).scalar()

6. Quản lý Transaction và Context Manager

Để đảm bảo an toàn dữ liệu và tránh rò rỉ kết nối, việc sử dụng Context Manager là một mô hình quản lý tốt.

from contextlib import contextmanager

@contextmanager
def session_scope():
    """Cung cấp một phạm vi giao dịch cho một loạt các thao tác."""
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Cách sử dụng
with session_scope() as session:
    user = Customer(fullname="Giao Dịch An Toàn", email="secure@example.com")
    session.add(user)

7. Tối ưu hóa hiệu năng

Trong các hệ thống lớn, cần lưu ý các kỹ thuật sau:

• Eager Loading: Sử dụng joinedload hoặc subqueryload để tránh vấn đề N+1 query khi truy cập các quan hệ.
• Connection Pooling: Điều chỉnh kích thước pool (pool_size) và thời gian timeout trong Engine để tối ưu hóa việc tái sử dụng kết nối.
• Index: Đảm bảo các cột thường xuyên dùng để lọc (filter) được đánh index trong Model định nghĩa.

from sqlalchemy.orm import joinedload

# Eager loading để lấy khách hàng kèm theo tất cả đơn hàng chỉ trong 1-2 câu lệnh SQL
customers_with_orders = db_session.query(Customer).options(joinedload(Customer.orders)).all()