SQLAlchemy/SQLModel多表join关联查询如何获取结果总行数

你当前写法返回的count全为1是符合SQL执行逻辑的：GROUP BY Transaction.id 会按交易ID拆分独立分组，count(Transaction.id) 统计的是单个分组内的行数，由于Transaction.id是分组主键，每个分组恰好对应1行数据，自然拿不到整个结果集的总行数。
直接移除group_by会报错、且统计结果不准的原因是多表关联存在表间对应关系，去掉分组后要么触发SQL语法错误（非聚合字段不在group by中），要么会把关联产生的重复关联行计入count，得到的数值远大于实际业务结果行数。

根据你的业务场景选对应方案即可：

方案1：分页场景（推荐，性能最好） #

绝大多数多表查询都是分页场景，需要同时返回当前页数据列表+符合筛选条件的总记录数。这种场景不要把count统计和业务字段写在同一个SELECT语句里，抽离公共查询逻辑复用即可，避免重复写关联、筛选条件：

from sqlalchemy import func, select
from sqlmodel import Session

# 抽离公共查询部分：只放多表关联、WHERE筛选条件，不要加字段选择、分组、排序、分页参数
base_query = (
    select(Transaction)
    .join(TransactionType)
    .join(xxx)
    # 此处追加所有业务筛选条件，例如：
    # .where(Transaction.status == 1)
    # .where(TransactionType.code.in_(['PAY', 'REFUND']))
)

# 1. 查询当前页列表数据
list_query = (
    base_query
    .add_columns(Transaction.id, TransactionType.name) # 追加你需要返回的业务字段
    .group_by(Transaction.id) # 和你原有分组逻辑保持一致
    .order_by(Transaction.create_time.desc())
    .offset((page_num - 1) * page_size)
    .limit(page_size)
)
data_list = session.exec(list_query).all()

# 2. 查询符合条件的总行数
# 将分组后的业务查询作为子查询，外层统计行数，完全规避group by对count的影响
count_query = select(func.count()).select_from(
    base_query.group_by(Transaction.id).subquery()
)
total = session.exec(count_query).one()

这个写法统计的总条数和实际列表返回的行数完全一致，且count查询不需要返回业务字段、不需要排序，性能比把count写在同一条语句里高很多。

方案2：非分页场景，单条查询返回行数据+总行数 #

如果不需要分页，想在每一行结果里直接拿到整个结果集的总行数，用窗口函数实现即可，不需要拆成两条查询：

from sqlalchemy import over

query = (
    select(
        Transaction.id,
        # 用over()定义窗口函数，不指定分区规则时默认统计整个结果集的总行数
        func.count(Transaction.id).over().label('total')
    )
    .join(TransactionType)
    .join(xxx)
    .group_by(Transaction.id)
)
result = session.exec(query).all()
# 返回的每一行数据里，total字段的值就是整个查询结果的总行数

注意：无论用哪种方案，count统计时的关联关系、筛选条件、分组规则必须和你查询列表数据的逻辑完全一致，否则会出现总条数和实际列表长度不匹配的问题。

内容的提问来源于stack exchange，提问作者johojojoj