Sử dụng SQLAlchemy ORM để Thao Tác Cơ Sở Dữ Liệu

SQLAlchemy là một trong những khung ORM (Object-Relational Mapping) phổ biến nhất của Python, cung cấp cách thức tương tác với cơ sở dữ liệu hiệu quả và linh hoạt. 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 với cơ sở dữ liệu.

Nội dung chính

  1. Cài đặt SQLAlchemy
  2. Các khái niệm cơ bản
  3. Kết nối đến cơ sở dữ liệu
  4. Định nghĩa mô hình dữ liệu
  5. Tạo bảng cơ sở dữ liệu
  6. Thao tác CRUD cơ bản
  7. Truy vấn dữ liệu
  8. Xử lý quan hệ
  9. Quản lý giao dịch
  10. Lời khuyên tốt nhất

Cài đặt

pip install sqlalchemy

Nếu bạn muốn kết nối đến các cơ sở dữ liệu cụ thể, hãy cài đặt các trình điều khiển tương ứng:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (đã được bao gồm trong thư viện chuẩn Python, không cần cài đặt thêm)

Các khái niệm cơ bản

  • Engine: Mô-đun kết nối cơ sở dữ liệu, chịu trách nhiệm giao tiếp với cơ sở dữ liệu.
  • Session: Phiên làm việc cơ sở dữ liệu, quản lý tất cả các hoạt động duy trì.
  • Model: Lớp mô hình dữ liệu, tương ứng với các bảng trong cơ sở dữ liệu.
  • Query: Đối tượng truy vấn, dùng để xây dựng và thực thi các truy vấn cơ sở dữ liệu.

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

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

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

# Ví dụ PostgreSQL
# engine = create_engine('postgresql://ten_dang_nhap:mat_khau@localhost:5432/ten_cua_so_du_lieu')

# Ví dụ MySQL
# engine = create_engine('mysql+mysqlconnector://ten_dang_nhap:mat_khau@localhost:3306/ten_cua_so_du_lieu')

# Tạo nhà máy phiên làm việc
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

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

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

# Tạo lớp cơ sở
Base = declarative_base()

class NguoiDung(Base):
    __tablename__ = 'nguoi_dung'
    
    id = Column(Integer, primary_key=True, index=True)
    ten = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    
    # Định nghĩa mối quan hệ một-nhiều
    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))
    tac_gia_id = Column(Integer, ForeignKey('nguoi_dung.id'))
    
    # Định nghĩa mối quan hệ nhiều-một
    tac_gia = relationship("NguoiDung", back_populates="bai_viet")
    
    # Định nghĩa mối quan hệ nhiều-nhiều (sử dụng bảng liên kết)
    the_tags = relationship("The", secondary="bai_viet_the", back_populates="bai_viet")

class The(Base):
    __tablename__ = 'the'
    
    id = Column(Integer, primary_key=True, index=True)
    ten = Column(String(30), unique=True, nullable=False)
    
    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)
class BaiVietThe(Base):
    __tablename__ = 'bai_viet_the'
    
    bai_viet_id = Column(Integer, ForeignKey('bai_viet.id'), primary_key=True)
    the_id = Column(Integer, ForeignKey('the.id'), primary_key=True)

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

# Tạo tất cả các bảng
Base.metadata.create_all(bind=engine)

# Xóa tất cả các bảng
# Base.metadata.drop_all(bind=engine)

Thao tác CRUD cơ bản

Tạo dữ liệu

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

# Tạo nhiều người dùng cùng lúc
session.add_all([
    NguoiDung(ten="Nguyễn Thị B", email="b.nguyenthi@example.com"),
    NguoiDung(ten="Trần Văn C", email="c.tranvan@example.com")
])
session.commit()

Đọc dữ liệu

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

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

# Lấy người dùng bằng ID
nguoi_dung = session.query(NguoiDung).get(1)

Cập nhật dữ liệu

# Truy vấn và cập nhật
nguoi_dung = session.query(NguoiDung).get(1)
nguoi_dung.ten = "Nguyễn Văn A1"
session.commit()

# Cập nhật nhiều người dùng cùng lúc
session.query(NguoiDung).filter(NguoiDung.ten.like("Nguyễn Văn %")).update({"ten": "Nguyễn Văn"}, synchronize_session=False)
session.commit()

Xóa dữ liệu

# Truy vấn và xóa
nguoi_dung = session.query(NguoiDung).get(1)
session.delete(nguoi_dung)
session.commit()

# Xóa nhiều người dùng cùng lúc
session.query(NguoiDung).filter(NguoiDung.ten == "Nguyễn Thị B").delete(synchronize_session=False)
session.commit()

Truy vấn dữ liệu

Truy vấn cơ bản

# Lấy tất cả các bản ghi
nguoi_dungs = session.query(NguoiDung).all()

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

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

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

# Bỏ qua một số bản ghi đầu
nguoi_dungs = session.query(NguoiDung).offset(5).limit(10).all()

Lọc truy vấn

from sqlalchemy import or_

# Lọc bằng giá trị bằng
nguoi_dung = session.query(NguoiDung).filter(NguoiDung.ten == "Nguyễn Văn A").first()

# Lọc bằng toán tử like
nguoi_dungs = session.query(NguoiDung).filter(NguoiDung.ten.like("Nguyễn Văn %")).all()

# Lọc bằng IN
nguoi_dungs = session.query(NguoiDung).filter(NguoiDung.ten.in_(["Nguyễn Văn A", "Nguyễn Thị B"])).all()

# Nhiều điều kiện lọc
nguoi_dungs = session.query(NguoiDung).filter(
    NguoiDung.ten == "Nguyễn Văn A", 
    NguoiDung.email.like("%@example.com")
).all()

# Điều kiện OR
nguoi_dungs = session.query(NguoiDung).filter(
    or_(NguoiDung.ten == "Nguyễn Văn A", NguoiDung.ten == "Nguyễn Thị B")
).all()

# Điều kiện khác
nguoi_dungs = session.query(NguoiDung).filter(NguoiDung.ten != "Nguyễn Văn A").all()

Truy vấn tổng hợp

from sqlalchemy import func

# Đếm số lượng
so_luong = session.query(NguoiDung).count()

# Đếm nhóm
so_bai_viet_nguoi_dung = session.query(
    NguoiDung.ten, 
    func.count(BaiViet.id)
).join(BaiViet).group_by(NguoiDung.ten).all()

# Tính tổng, trung bình, v.v.
trung_binh_id = session.query(func.avg(NguoiDung.id)).scalar()

Truy vấn liên kết

# Liên kết nội
ket_quas = session.query(NguoiDung, BaiViet).join(BaiViet).filter(BaiViet.tieu_de.like("%Python%")).all()

# Liên kết bên trái
ket_quas = session.query(NguoiDung, BaiViet).outerjoin(BaiViet).all()

# Liên kết với điều kiện cụ thể
ket_quas = session.query(NguoiDung, BaiViet).join(BaiViet, NguoiDung.id == BaiViet.tac_gia_id).all()

Xử lý quan hệ

# Tạo đối tượng có quan hệ
nguoi_dung = NguoiDung(ten="Nguyễn Văn D", email="d.nguyenvan@example.com")
bai_viet = BaiViet(tieu_de="Bài viết đầu tiên", noi_dung="Xin chào!", tac_gia=nguoi_dung)
session.add(bai_viet)
session.commit()

# Truy cập thông qua quan hệ
print(f"Bài viết '{bai_viet.tieu_de}' của tác giả {bai_viet.tac_gia.ten}")
print(f"Tất cả bài viết của người dùng {nguoi_dung.ten}:")
for bv in nguoi_dung.bai_viet:
    print(f"  - {bv.tieu_de}")

# Xử lý quan hệ nhiều-nhiều
the_python = The(ten="Python")
the_sqlalchemy = The(ten="SQLAlchemy")

bai_viet.the_tags.append(the_python)
bai_viet.the_tags.append(the_sqlalchemy)
session.commit()

print(f"Các thẻ của bài viết '{bai_viet.tieu_de}':")
for the in bai_viet.the_tags:
    print(f"  - {the.ten}")

Quản lý giao dịch

# Giao dịch tự động commit
try:
    nguoi_dung = NguoiDung(ten="Người dùng thử nghiệm", email="thuchien@example.com")
    session.add(nguoi_dung)
    session.commit()
except Exception as e:
    session.rollback()
    print(f"Lỗi xảy ra: {e}")

# Sử dụng quản lý ngữ cảnh giao dịch
from sqlalchemy.orm import Session

def them_nguoi_dung(session: Session, ten: str, email: str):
    try:
        nguoi_dung = NguoiDung(ten=ten, email=email)
        session.add(nguoi_dung)
        session.commit()
        return nguoi_dung
    except:
        session.rollback()
        raise

# Giao dịch lồng nhau
with session.begin_nested():
    nguoi_dung = NguoiDung(ten="Người dùng giao dịch", email="giaodich@example.com")
    session.add(nguoi_dung)

# Điểm lưu
diem_luu = session.begin_nested()
try:
    nguoi_dung = NguoiDung(ten="Người dùng điểm lưu", email="diemluu@example.com")
    session.add(nguoi_dung)
    diem_luu.commit()
except:
    diem_luu.rollback()

Lời khuyên tốt nhất

  1. Quản lý phiên làm việc: Tạo phiên làm việc mới cho mỗi yêu cầu và đóng lại sau khi hoàn thành.
  2. Xử lý ngoại lệ: Luôn xử lý ngoại lệ và hoàn tác giao dịch khi cần thiết.
  3. Tải chậm: Chú ý vấn đề N+1 truy vấn, tối ưu hóa bằng cách tải sớm.
  4. Bộ nhớ đệm kết nối: Cấu hình kích thước và thời gian chờ đợi phù hợp cho bộ nhớ đệm kết nối.
  5. Đảm bảo tính toàn vẹn dữ liệu: Kiểm tra tính hợp lệ của dữ liệu ở tầng mô hình hoặc ứng dụng.
# Sử dụng quản lý ngữ cảnh để quản lý phiên làm việc
from contextlib import contextmanager

@contextmanager
def lay_db():
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

# Ví dụ sử dụng
with lay_db() as db:
    nguoi_dung = NguoiDung(ten="Người dùng ngữ cảnh", email="ngonngu_hoc@example.com")
    db.add(nguoi_dung)

Thẻ: sqlalchemy orm python Cơ_sở_dữ_liệu

Đăng vào ngày 6 tháng 7 lúc 17:26

Thẻ Phổ Biến

©2026 Thành phố Cuồng loạn