Thiết lập môi trường và phụ thuộc

Trước khi triển khai, hãy cài đặt thư viện cốt lõi cùng trình điều khiển tương ứng với hệ quản trị cơ sở dữ liệu bạn dự định sử dụng:

pip install sqlalchemy
pip install psycopg2-binary  # Hỗ trợ PostgreSQL
pip install mysqlclient      # Hỗ trợ MySQL

Các thành phần nền tảng

• Engine: Cổng kết nối vật lý, xử lý nhóm kết nối (connection pooling) và giao tiếp trực tiếp với hệ quản trị.
• Session: Không gian làm việc tạm thời, theo dõi trạng thái đối tượng và quyết định thời điểm đồng bộ xuống đĩa.
• Model: Lớp Python ánh xạ 1:1 với cấu trúc bảng và cột trong cơ sở dữ liệu.
• Query/Select: Công cụ xây dựng câu lệnh truy vấn an toàn kiểu, tránh lỗi cú pháp SQL thô.

Khởi tạo kênh kết nối

Thiết lập engine và session factory để đảm bảo tính nhất quán và tái sử dụng tài nguyên:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Chuỗi kết nối SQLite
db_uri = "sqlite:///inventory_db.sqlite"
engine_pool = create_engine(db_uri, echo=False)

# Factory tạo phiên làm việc
SessionFactory = sessionmaker(bind=engine_pool, autocommit=False, autoflush=False)

# Khởi tạo phiên cụ thể
active_session = SessionFactory()

Xây dựng mô hình thực thể

Sử dụng lớp nền tảng để định nghĩa ánh xạ quan hệ và ràng buộc:

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

SchemaBase = declarative_base()

class Product(SchemaBase):
    __tablename__ = "products"

    product_id = Column(Integer, primary_key=True, autoincrement=True)
    sku = Column(String(20), unique=True, nullable=False)
    name = Column(String(100))
    price = Column(Float)
    warehouse_id = Column(Integer, ForeignKey("warehouses.warehouse_id"))

    storage_location = relationship("Warehouse", back_populates="items")
    categories = relationship(
        "Category",
        secondary="product_category_link",
        back_populates="assigned_products"
    )

class Warehouse(SchemaBase):
    __tablename__ = "warehouses"

    warehouse_id = Column(Integer, primary_key=True)
    location_code = Column(String(30), unique=True)
    capacity = Column(Integer)

    items = relationship("Product", back_populates="storage_location")

class Category(SchemaBase):
    __tablename__ = "categories"

    category_id = Column(Integer, primary_key=True)
    tag = Column(String(50), unique=True)

    assigned_products = relationship(
        "Product",
        secondary="product_category_link",
        back_populates="categories"
    )

# Bảng ánh xạ nhiều-nhiều
product_category_link = Table(
    "product_category_link",
    SchemaBase.metadata,
    Column("prod_ref", Integer, ForeignKey("products.product_id"), primary_key=True),
    Column("cat_ref", Integer, ForeignKey("categories.category_id"), primary_key=True)
)

Đồng bộ cấu trúc bảng

# Sinh bảng dựa trên mô hình đã khai báo
SchemaBase.metadata.create_all(bind=engine_pool)

# Xóa toàn bộ lược đồ (chỉ dùng trong môi trường thử nghiệm)
# SchemaBase.metadata.drop_all(bind=engine_pool)

Thao tác ghi đọc và sửa xóa

Thêm bản ghi mới

# Chèn đơn lẻ
new_item = Product(sku="A001", name="Bàn phím cơ", price=150.0)
active_session.add(new_item)
active_session.commit()

# Chèn hàng loạt
batch_items = [
    Product(sku="A002", name="Chuột không dây", price=45.0),
    Product(sku="A003", name="Màn hình 24 inch", price=220.0)
]
active_session.add_all(batch_items)
active_session.commit()

Truy xuất dữ liệu

# Lấy toàn bộ danh sách
all_items = active_session.query(Product).all()

# Lấy bản ghi đầu tiên trong kết quả
top_item = active_session.query(Product).first()

# Truy vấn theo khóa chính
target = active_session.get(Product, 1)

Cập nhật thông tin

# Sửa trực tiếp đối tượng
record = active_session.get(Product, 1)
record.price = 135.5
active_session.commit()

# Cập nhật theo điều kiện
active_session.query(Product).filter(Product.price > 200).update(
    {"price": 199.99},
    synchronize_session="fetch"
)
active_session.commit()

Xóa dữ liệu

# Xóa từng thực thể
to_remove = active_session.get(Product, 2)
active_session.delete(to_remove)
active_session.commit()

# Xóa theo bộ lọc
active_session.query(Product).filter(Product.sku == "TEMP_001").delete()
active_session.commit()

Kỹ thuật truy vấn nâng cao

Sắp xếp và phân trang

# Sắp xếp giảm dần theo giá
sorted_inventory = active_session.query(Product).order_by(Product.price.desc()).all()

# Phân trang: bỏ qua 3 bản ghi, lấy 5 bản ghi tiếp theo
paged_data = active_session.query(Product).offset(3).limit(5).all()

Bộ lọc phức hợp

from sqlalchemy import or_, and_

# Lọc trùng khớp chính xác
exact = active_session.query(Product).filter(Product.sku == "A001").first()

# Tìm kiếm theo mẫu ký tự
pattern_match = active_session.query(Product).filter(Product.name.like("%màn%")).all()

# Điều kiện kết hợp
multi_cond = active_session.query(Product).filter(
    and_(
        Product.price >= 50,
        Product.price <= 250
    )
).all()

# Toán tử OR
or_filter = active_session.query(Product).filter(
    or_(Product.name.contains("chuột"), Product.sku.startswith("B"))
).all()

Thống kê và nhóm

from sqlalchemy import func

# Đếm tổng số mặt hàng
total_products = active_session.query(Product).count()

# Nhóm theo kho và đếm số lượng
stock_stats = active_session.query(
    Warehouse.location_code,
    func.count(Product.product_id)
).join(Warehouse.items).group_by(Warehouse.location_code).all()

# Tính giá trị trung bình
avg_cost = active_session.query(func.avg(Product.price)).scalar()

Truy vấn kết hợp (Join)

# Inner Join
joined_records = active_session.query(Product, Warehouse).join(Warehouse).filter(Warehouse.location_code == "HANOI_01").all()

# Left Outer Join
left_joined = active_session.query(Product, Warehouse).outerjoin(Warehouse).all()

# Join tường minh với điều kiện
explicit_join = active_session.query(Product, Warehouse).join(
    Warehouse, Product.warehouse_id == Warehouse.warehouse_id
).all()

Xử lý quan hệ thực thể

# Gán vị trí lưu kho
storage = Warehouse(warehouse_id=10, location_code="SG_HCM_02", capacity=5000)
item = Product(sku="C001", name="Ổ cứng SSD", price=80.0, storage_location=storage)
active_session.add(item)
active_session.commit()

# Truy xuất ngược qua relationship
print(f"Mặt hàng: {item.name} | Kho: {item.storage_location.location_code}")

# Gán danh mục (nhiều-nhiều)
electronics = Category(category_id=1, tag="Điện tử")
storage_device = Category(category_id=2, tag="Lưu trữ")
item.categories.extend([electronics, storage_device])
active_session.commit()

# Duyệt danh mục
for cat in item.categories:
    print(f"  - {cat.tag}")

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

# Xử lý thủ công với rollback
try:
    temp_prod = Product(sku="TEMP_X", name="Kiểm tra", price=0.0)
    active_session.add(temp_prod)
    active_session.commit()
except Exception as err:
    active_session.rollback()
    print(f"Giao dịch thất bại: {err}")

# Hàm bao đóng quản lý giao dịch
def register_product(sess, sku_val: str, name_val: str, price_val: float):
    try:
        prod = Product(sku=sku_val, name=name_val, price=price_val)
        sess.add(prod)
        sess.commit()
        return prod
    except:
        sess.rollback()
        raise

# Sử dụng savepoint (giao dịch lồng)
with active_session.begin_nested():
    nested_prod = Product(sku="NEST_01", name="Lồng giao dịch", price=10.0)
    active_session.add(nested_prod)

# Tạo và kiểm soát điểm khôi phục trực tiếp
sp = active_session.begin_nested()
try:
    sp_prod = Product(sku="SP_01", name="Điểm lưu trữ", price=20.0)
    active_session.add(sp_prod)
    sp.commit()
except:
    sp.rollback()

Quy chuẩn tối ưu hiệu năng

• Vòng đời Session: Luôn tạo phiên mới cho mỗi request và đóng ngay sau khi xử lý xong.
• Xử lý ngoại lệ: Bao bọc commit() trong try...except để đảm bảo gọi rollback() khi phát sinh lỗi.
• Giảm truy vấn thừa: Tránh hiện tượng N+1 bằng cách dùng joinedload hoặc subqueryload thay vì lazy loading mặc định.
• Cấu hình Pool: Tinh chỉnh pool_size, max_overflow và pool_timeout phù hợp với lưu lượng thực tế.
• Xác thực dữ liệu: Kết hợp Pydantic hoặc Marshmallow để kiểm tra tính toàn vẹn trước khi ánh xạ vào ORM.

from contextlib import contextmanager

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

# Tiêu thụ an toàn
with open_db_scope() as db:
    entity = Product(sku="CTX_01", name="Quản lý ngữ cảnh", price=99.0)
    db.add(entity)