Python 数据库进阶：SQLAlchemy ORM 高级查询、性能优化、事务管理

小飞兽 Python 12 次阅读 2026-07-24

Introduction

SQLAlchemy 是 Python 最强大的 ORM 库。本文讲解关系定义、一对多/多对多查询、懒加载vs急加载、批量操作、性能优化（索引、查询优化）、事务管理和连接池配置。

---

模型关系定义

from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, Boolean
from sqlalchemy.orm import relationship, declarative_base
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True)
    posts = relationship('Post', back_populates='author', cascade='all, delete-orphan')
    tags = relationship('Tag', secondary=user_tags, back_populates='users')

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text)
    author_id = Column(Integer, ForeignKey('users.id'))
    author = relationship('User', back_populates='posts')
    comments = relationship('Comment', back_populates='post', cascade='all, delete-orphan')
    created_at = Column(DateTime, default=datetime.utcnow)
    is_published = Column(Boolean, default=False)

class Comment(Base):
    __tablename__ = 'comments'
    id = Column(Integer, primary_key=True)
    content = Column(Text)
    post_id = Column(Integer, ForeignKey('posts.id'))
    post = relationship('Post', back_populates='comments')
    author_id = Column(Integer, ForeignKey('users.id'))

基本 CRUD

from sqlalchemy.orm import Session

session = Session(bind=engine)

# 创建
user = User(username='zhangsan', email='zhangsan@example.com')
session.add(user)
session.commit()

# 读取
user = session.query(User).filter_by(username='zhangsan').first()
post = session.query(Post).get(1)  # 按主键

# 更新
user.email = 'new@example.com'
session.commit()

# 删除
session.delete(user)
session.commit()

高级查询

from sqlalchemy import func, and_, or_, desc

# 条件查询
posts = session.query(Post).filter(
    and_(Post.is_published == True, Post.created_at >= '2026-01-01')
).order_by(desc(Post.created_at)).all()

# 模糊搜索
results = session.query(Post).filter(
    Post.title.like('%Python%')
).all()

# 聚合统计
stats = session.query(
    User.username,
    func.count(Post.id).label('post_count')
).join(Post).group_by(User.username).having(
    func.count(Post.id) >= 5
).all()

# 子查询
subq = session.query(func.max(Post.created_at)).scalar()
latest_posts = session.query(Post).filter(
    Post.created_at == subq
).all()

# 分页
page = 2; page_size = 20
posts = session.query(Post).limit(page_size).offset((page-1)*page_size).all()

关系加载策略

# 懒加载（默认，会 N+1 问题）
posts = session.query(Post).limit(10).all()
for p in posts:
    print(p.author.username)  # 每次都查数据库！

# 急加载（解决 N+1）
from sqlalchemy.orm import joinedload, selectinload

posts = session.query(Post).options(
    joinedload(Post.author)
).limit(10).all()
for p in posts:
    print(p.author.username)  # JOIN 一次查出

# selectinload（适合一对多）
posts = session.query(Post).options(
    selectinload(Post.comments)
).limit(10).all()

批量操作（性能优化）

# 批量插入
session.bulk_insert_mappings(User, [
    {'username': f'user{i}', 'email': f'user{i}@example.com'}
    for i in range(1000)
])
session.commit()

# 批量更新
session.query(User).filter(User.username.like('test%')).update(
    {'is_active': False}, synchronize_session='fetch'
)
session.commit()

# 批量删除
session.query(Post).filter(Post.is_published == False).delete(synchronize_session=False)
session.commit()

事务管理

try:
    with session.begin():
        user = User(username='newuser', email='new@example.com')
        session.add(user)
        post = Post(title='First Post', content='...', author_id=user.id)
        session.add(post)
        # 两者在同一个事务中
except Exception as e:
    session.rollback()  # 事务回滚
    print(f'事务失败: {e}')

性能优化技巧

# 1. 创建索引
from sqlalchemy import Index
Index('idx_post_created', Post.created_at)
Index('idx_post_author_published', 'author_id', 'is_published')

# 2. 查询时只选需要的列
posts = session.query(Post.id, Post.title).filter(...).all()

# 3. 使用 exists 而非 IN
from sqlalchemy import exists
has_posts = session.query(User).filter(
    exists().where(Post.author_id == User.id)
).all()

# 4. 连接池配置
engine = create_engine('mysql+pymysql://user:pass@host/db',
    pool_size=10, max_overflow=20, pool_recycle=3600)

常见问题

Q1: N+1 查询问题怎么解决？用 joinedload() 或 selectinload() 急加载关联数据。

Q2: session 和 engine 区别？engine 是数据库连接工厂，session 是 ORM 操作的工作单元（Unit of Work）。

Q3: 如何处理数据库连接超时？配置 pool_recycle=3600（1小时刷新连接）和 pool_pre_ping=True（每次使用前检测连接是否有效）。