第五章：高级查询

JOIN 查询

内连接

# JOIN 查询
result = session.query(User, Post).join(Post).all()

for user, post in result:
    print(f"{user.name}: {post.title}")

# 指定条件
result = session.query(User, Post).join(
    Post, User.id == Post.user_id
).all()

左连接

from sqlalchemy.orm import outerjoin

# 左连接
result = session.query(User, Post).outerjoin(
    Post, User.id == Post.user_id
).all()

for user, post in result:
    if post:
        print(f"{user.name}: {post.title}")
    else:
        print(f"{user.name}: No posts")

子查询

标量子查询

from sqlalchemy import scalar_subquery

# 查询每个用户的文章数
subquery = session.query(
    func.count(Post.id)
).filter(Post.user_id == User.id).scalar_subquery()

result = session.query(
    User.name,
    subquery.label('post_count')
).all()

EXISTS 子查询

from sqlalchemy import exists

# 查询有文章的用户
result = session.query(User).filter(
    exists().where(Post.user_id == User.id)
).all()

关联加载

N+1 问题

# 问题：N+1 查询
users = session.query(User).all()
for user in users:
    print(user.posts)  # 每次都查询

预加载

from sqlalchemy.orm import joinedload, selectinload

# joinedload（JOIN 加载）
users = session.query(User).options(
    joinedload(User.posts)
).all()

# selectinload（IN 加载）
users = session.query(User).options(
    selectinload(User.posts)
).all()

延迟加载

from sqlalchemy.orm import lazyload

# 延迟加载
users = session.query(User).options(
    lazyload(User.posts)
).all()

小结

高级查询要点：

• JOIN 查询：join()、outerjoin()
• 子查询：scalar_subquery、exists
• 关联加载：joinedload、selectinload、lazyload

下一章我们将学习事务管理。