5.1 连接查询与聚合统计

本节解决两类高频需求：跨表查询（JOIN） 和 统计分析（COUNT/SUM/AVG + GROUP BY）。学完你就能写出"每个用户发了几篇文章"这种报表级查询。

示例沿用 4.1 节的 User / Post 模型。

一、join：跨表过滤

需求："查出张三写的所有文章"。文章表里只有 user_id 没有名字，必须联合两张表查询：

from sqlalchemy import select

stmt = (
    select(Post)
    .join(Post.author)              # 沿着关系 JOIN 到 users 表
    .where(User.name == "张三")
)
posts = session.scalars(stmt).all()

生成的 SQL：

SELECT posts.*
FROM posts JOIN users ON users.id = posts.user_id
WHERE users.name = '张三'

推荐沿着 relationship join（join(Post.author)），SQLAlchemy 自动知道 ON 条件。也可以写目标类 join(User)（会根据外键自动推断），效果相同。

join 之后可以随意引用两张表的列

# 查文章标题 + 作者名（两张表的列混合查）
stmt = (
    select(Post.title, User.name)
    .join(Post.author)
    .where(User.name.in_(["张三", "李四"]))
    .order_by(User.name)
)
for title, author_name in session.execute(stmt):
    print(f"{author_name}：《{title}》")

join vs joinedload，别搞混！

	.join(...)	.options(joinedload(...))
目的	过滤/选择：让 where 能用另一张表的列	预加载：把关系数据装进对象属性
影响返回内容	不装载关联对象	装载 post.author

需求是"按作者名过滤并且之后要访问 post.author"？两个都要写：

stmt = (
    select(Post)
    .join(Post.author)
    .where(User.name == "张三")
    .options(joinedload(Post.author))
)

二、外连接：outerjoin

join 是内连接——"一篇文章都没有的用户"会被丢掉。想保留左表所有行，用 LEFT OUTER JOIN：

# 所有用户 + 各自的文章（没文章的用户也要出现，posts 列为 NULL）
stmt = select(User.name, Post.title).outerjoin(User.posts)
for name, title in session.execute(stmt):
    print(name, title)     # 没文章的用户：('王五', None)

三、聚合函数：func

func 是 SQL 函数的万能入口，func.任意名字 会原样变成 SQL 函数：

from sqlalchemy import func, select

# 用户总数
session.scalar(select(func.count()).select_from(User))

# 最大/最小/平均/总和
session.scalar(select(func.max(User.age)))
session.scalar(select(func.avg(User.age)))
session.scalar(select(func.sum(User.age)))
session.scalar(select(func.min(User.created_at)))

# 非重复计数：有多少个不同的城市
session.scalar(select(func.count(User.city.distinct())))

四、group_by：分组统计

"每个城市有多少用户"：

stmt = (
    select(User.city, func.count(User.id).label("user_count"))
    .group_by(User.city)
    .order_by(func.count(User.id).desc())
)
for city, count in session.execute(stmt):
    print(f"{city}: {count} 人")

SELECT users.city, count(users.id) AS user_count
FROM users GROUP BY users.city
ORDER BY count(users.id) DESC

• .label("别名") 给计算列起名，结果里可以用 row.user_count 访问
• GROUP BY 铁律：select 里的非聚合列必须出现在 group_by 里

having：对分组结果过滤

where 过滤的是行，having 过滤的是分组之后的组：

# 只看用户数超过 2 人的城市
stmt = (
    select(User.city, func.count(User.id).label("cnt"))
    .group_by(User.city)
    .having(func.count(User.id) > 2)
)

五、组合拳：join + group_by

报表级需求："每个用户发了多少篇文章，按文章数倒序"：

stmt = (
    select(User.name, func.count(Post.id).label("post_count"))
    .outerjoin(User.posts)                 # outerjoin 保证 0 篇的用户也统计到
    .group_by(User.id)
    .order_by(func.count(Post.id).desc())
)
for name, post_count in session.execute(stmt):
    print(f"{name} 发表了 {post_count} 篇")

SELECT users.name, count(posts.id) AS post_count
FROM users LEFT OUTER JOIN posts ON users.id = posts.user_id
GROUP BY users.id
ORDER BY count(posts.id) DESC

六、子查询初步

"查文章数超过平均值的用户"这类需求需要子查询。入门阶段掌握最常用的一种——把统计结果当临时表再 join：

# 每个用户的文章数（子查询）
post_count_sq = (
    select(Post.user_id, func.count(Post.id).label("cnt"))
    .group_by(Post.user_id)
    .subquery()
)

# 主查询：join 子查询，筛出文章数 >= 2 的用户
stmt = (
    select(User.name, post_count_sq.c.cnt)
    .join(post_count_sq, User.id == post_count_sq.c.user_id)
    .where(post_count_sq.c.cnt >= 2)
)

• .subquery() 把一条 select 变成"临时表"
• 访问子查询的列用 .c.列名（c = columns）
• join 子查询时要手动写 ON 条件

子查询是进阶中的进阶，第一遍学习看懂示例即可，用到时回来抄模板。

七、常用 SQL 函数速查

func.count(X)、func.sum(X)、func.avg(X)、func.max(X)、func.min(X)   # 聚合
func.now()                        # 当前时间（server_default 常用）
func.coalesce(X, 默认值)          # X 为 NULL 时用默认值（统计时救场神器）
func.length(User.name)            # 字符串长度
func.lower(X) / func.upper(X)     # 大小写转换
func.date(User.created_at)        # 取日期部分（按天分组统计常用）

例：按天统计注册量

stmt = (
    select(func.date(User.created_at).label("day"), func.count())
    .group_by(func.date(User.created_at))
    .order_by("day")
)

📝 本节小结

• 跨表过滤用 .join(关系)；要保留无匹配行用 .outerjoin()
• join 管过滤、joinedload 管装载，需求重叠时两个一起写
• 聚合走 func.count/sum/avg/...，配 group_by 分组、having 过滤组、.label() 起别名
• "每个X有多少Y" = outerjoin + group_by 模板
• 子查询：.subquery() + .c.列名，先会抄模板

下一节 → 5.2 事务处理