Giới thiệu SQLAlchemy ORM: Thao tác cơ sở dữ liệu hiệu quả trong Python

SQLAlchemy là một trong những framework ORM phổ biến nhất của Python, cung cấp cách thao tác cơ sở dữ liệu hiệu quả và linh hoạt. Bài viết này hướng dẫn sử dụng SQLAlchemy ORM để làm việc với cơ sở dữ liệu.

Mục lục

Cài đặt SQLAlchemy
Khái niệm cốt lõi
Kết nối cơ sở dữ liệu
Định nghĩa mô hình dữ liệu
Tạo bảng cơ sở dữ liệu
CRUD cơ bản
Truy vấn dữ liệu
Thao tác quan hệ
Quản lý giao dịch
Thực hành tốt

Cài đặt

bash

pip install sqlalchemy

Đối với cơ sở dữ liệu cụ thể, cần cài thêm driver:

bash

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (đã bao gồm trong Python)

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

Engine: Máy phát động kết nối cơ sở dữ liệu
Session: Phiên làm việc quản lý thao tác dữ liệu
Model: Lớp mô hình dữ liệu ánh xạ sang bảng
Query: Đối tượng truy vấn xây dựng và thực thi lệnh SQL

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

python

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Tạo engine kết nối
engine = create_engine('sqlite:///example.db', echo=True)

# Tạo nhà máy phiên làm việc
DBSession = sessionmaker(bind=engine)

# Tạo phiên làm việc
session = DBSession()

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

python

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

Base = declarative_base()

class NguoiDung(Base):
    __tablename__ = 'nguoidungs'
    
    id = Column(Integer, primary_key=True)
    ten = Column(String(50))
    email = Column(String(100), unique=True)
    
    bai_viet = relationship("BaiViet", back_populates="tac_gia")
    
class BaiViet(Base):
    __tablename__ = 'baiviet'
    
    id = Column(Integer, primary_key=True)
    tieu_de = Column(String(100))
    noi_dung = Column(String(500))
    tac_gia_id = Column(Integer, ForeignKey('nguoidungs.id'))
    
    tac_gia = relationship("NguoiDung", back_populates="bai_viet")
    
    the_loai = relationship("TheLoai", secondary="baiviet_theloai", back_populates="bai_viet")

class TheLoai(Base):
    __tablename__ = 'theloai'
    
    id = Column(Integer, primary_key=True)
    ten_the_loai = Column(String(30), unique=True)
    
    bai_viet = relationship("BaiViet", secondary="baiviet_theloai", back_populates="the_loai")

class BaiVietTheLoai(Base):
    __tablename__ = 'baiviet_theloai'
    
    bai_viet_id = Column(Integer, ForeignKey('baiviet.id'), primary_key=True)
    the_loai_id = Column(Integer, ForeignKey('theloai.id'), primary_key=True)

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

python

Base.metadata.create_all(engine)

CRUD cơ bản

Tạo dữ liệu

python

# Tạo người dùng mới
nguoi_dung_moi = NguoiDung(ten="Nguyen Van A", email="nva@example.com")
session.add(nguoi_dung_moi)
session.commit()

# Tạo nhiều người dùng
session.add_all([
    NguoiDung(ten="Tran Thi B", email="ttb@example.com"),
    NguoiDung(ten="Le Van C", email="lvc@example.com")
])
session.commit()

Đọc dữ liệu

python

# Lấy tất cả người dùng
danh_sach_nguoi_dung = session.query(NguoiDung).all()

# Lấy người dùng đầu tiên
nguoi_dung_dau_tien = session.query(NguoiDung).first()

# Lấy theo ID
nguoi_dung = session.query(NguoiDung).get(1)

Cập nhật dữ liệu

python

# Cập nhật thông tin
nguoi_dung = session.query(NguoiDung).get(1)
nguoi_dung.ten = "Nguyen Van An"
session.commit()

# Cập nhật hàng loạt
session.query(NguoiDung).filter(NguoiDung.ten.like("Nguyen%")).update({"ten": "Nguyen"}, synchronize_session=False)
session.commit()

Xóa dữ liệu

python

# Xóa người dùng
session.delete(session.query(NguoiDung).get(1))
session.commit()

# Xóa theo điều kiện
session.query(NguoiDung).filter(NguoiDung.ten == "Tran Thi B").delete(synchronize_session=False)
session.commit()

Truy vấn dữ liệu

Truy vấn cơ bản

python

# Lấy toàn bộ bản ghi
danh_sach = session.query(NguoiDung).all()

# Lấy trường cụ thể
ten = session.query(NguoiDung.ten).all()

# Sắp xếp
danh_sach = session.query(NguoiDung).order_by(NguoiDung.ten.desc()).all()

# Giới hạn kết quả
danh_sach = session.query(NguoiDung).limit(5).all()

# Phân trang
danh_sach = session.query(NguoiDung).offset(10).limit(5).all()

Truy vấn điều kiện

python

from sqlalchemy import or_

# Điều kiện bằng
kq = session.query(NguoiDung).filter(NguoiDung.ten == "Nguyen Van A").first()

# Tìm kiếm mờ
kq = session.query(NguoiDung).filter(NguoiDung.ten.like("Nguyen%")).all()

# IN clause
kq = session.query(NguoiDung).filter(NguoiDung.ten.in_(["Nguyen Van A", "Tran Thi B"])).all()

# Nhiều điều kiện
kq = session.query(NguoiDung).filter(
    NguoiDung.ten == "Nguyen Van A", 
    NguoiDung.email.like("%@example.com")
).all()

# Hoặc điều kiện
kq = session.query(NguoiDung).filter(
    or_(NguoiDung.ten == "Nguyen Van A", NguoiDung.ten == "Tran Thi B")
).all()

# Khác
kq = session.query(NguoiDung).filter(NguoiDung.ten != "Nguyen Van A").all()

Truy vấn tổng hợp

python

from sqlalchemy import func

# Đếm bản ghi
so_luong = session.query(NguoiDung).count()

# Nhóm và đếm
ket_qua = session.query(
    NguoiDung.ten, 
    func.count(BaiViet.id)
).join(BaiViet).group_by(NguoiDung.ten).all()

# Tính toán
gia_tri_trung_binh = session.query(func.avg(NguoiDung.id)).scalar()

Truy vấn join

python

# Inner join
kq = session.query(NguoiDung, BaiViet).join(BaiViet).filter(BaiViet.tieu_de.like("%Python%")).all()

# Left outer join
kq = session.query(NguoiDung, BaiViet).outerjoin(BaiViet).all()

# Join với điều kiện
kq = session.query(NguoiDung, BaiViet).join(BaiViet, NguoiDung.id == BaiViet.tac_gia_id).all()

Thao tác quan hệ

python

# Tạo đối tượng có quan hệ
nd = NguoiDung(ten="Pham Van D", email="pvd@example.com")
bv = BaiViet(tieu_de="Bài viết đầu tiên", noi_dung="Xin chào", tac_gia=nd)
session.add(bv)
session.commit()

# Truy cập quan hệ
print(f"Bài '{bv.tieu_de}' do {bv.tac_gia.ten} viết")
print(f"Tất cả bài của {nd.ten}:")
for bai in nd.bai_viet:
    print(f"- {bai.tieu_de}")

# Quan hệ nhiều-nhiều
tl1 = TheLoai(ten_the_loai="Python")
tl2 = TheLoai(ten_the_loai="SQL")

bv.the_loai.extend([tl1, tl2])
session.commit()

print(f"Thể loại bài '{bv.tieu_de}':")
for theloai in bv.the_loai:
    print(f"- {theloai.ten_the_loai}")

Quản lý giao dịch

python

# Xử lý giao dịch
try:
    nd = NguoiDung(ten="Test User", email="test@example.com")
    session.add(nd)
    session.commit()
except:
    session.rollback()
    print("Lỗi xảy ra")

# Quản lý giao dịch bằng context
def tao_nguoi_dung(session, ten, email):
    try:
        session.add(NguoiDung(ten=ten, email=email))
        session.commit()
    except:
        session.rollback()
        raise

# Giao dịch lồng
with session.begin_nested():
    nd = NguoiDung(ten="Nested User", email="nested@example.com")
    session.add(nd)

# Savepoint
sp = session.begin_nested()
try:
    nd = NguoiDung(ten="Savepoint User", email="savepoint@example.com")
    session.add(nd)
    sp.commit()
except:
    sp.rollback()

Thực hành tốt

Quản lý phiên: Tạo phiên mới cho mỗi yêu cầu
Xử lý lỗi: Luôn rollback khi có exception
Tối ưu truy vấn: Tránh N+1 queries bằng eager loading
Cấu hình pool: Đặt kích thước và timeout phù hợp
Kiểm tra dữ liệu: Xác thực dữ liệu ở lớp mô hình

python

# Quản lý phiên bằng context manager
from contextlib import contextmanager

@contextmanager
def get_db():
    db = DBSession()
    try:
        yield db
        db.commit()
    except:
        db.rollback()
        raise
    finally:
        db.close()

# Sử dụng
with get_db() as db:
    db.add(NguoiDung(ten="Context User", email="context@example.com"))

Thẻ: sqlalchemy orm python Database Object Relational Mapping

Đăng vào ngày 21 tháng 6 lúc 22:41

Thành phố Cuồng loạn

Giới thiệu SQLAlchemy ORM: Thao tác cơ sở dữ liệu hiệu quả trong Python

Mục lục

Cài đặt

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

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

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

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

CRUD cơ bản

Tạo dữ liệu

Đọc dữ liệu

Cập nhật dữ liệu

Xóa dữ liệu

Truy vấn dữ liệu

Truy vấn cơ bản

Truy vấn điều kiện

Truy vấn tổng hợp

Truy vấn join

Thao tác quan hệ

Quản lý giao dịch

Thực hành tốt

Thẻ Phổ Biến