升级SQLAlchemy后如何不用原生SQL高效获取ORM查询结果行数

问题原因 #

AttributeError: 'LegacyCursorResult' object has no attribute '_saved_cursor'

SQLAlchemy 1.4及以上版本重构了结果集实现，_saved_cursor、_result这类单下划线开头的属性是框架内部私有属性，原本就不推荐业务代码直接调用，版本升级后直接废弃移除导致报错。
你目前使用的遍历计数方案本质是把所有符合条件的行全部拉取到本地再计数，数据量大时IO、内存开销都很高，性能极差。

最优解决方案：数据库端聚合计数 #

直接使用SQLAlchemy提供的func.count() API构造聚合查询，计数逻辑完全在数据库端执行，最终仅返回单个计数值，不需要拉取任何业务行数据，是性能最高的实现方式，全程不需要写原生SQL。

适配现有代码的实现如下：

from sqlalchemy import func

def get_clients(db, status):
    clients = Table("clients", db.metadata, autoload=True)
    # 构造count聚合查询，无主键表可替换为func.count(1)
    qry = db.select(func.count(clients.c.id)).where(clients.c.status == status)
    # 执行后直接取单个计数结果
    rowcount = db.execute(qry).scalar()
    return rowcount

备选方案：已拿到结果集时的简化写法 #

如果你已经执行过查询拿到了结果集，不想再发一次计数请求，可以直接用len(res.all())替代遍历计数，写法更简洁，但性能和遍历完全一致，仅适合小数据量场景：

def get_clients(db, status):
    clients = Table("clients", db.metadata, autoload=True)
    qry = clients.select().where(clients.c.status == status)
    res = qry.execute()
    rowcount = len(res.all())
    return rowcount

注意：不要使用res.rowcount属性获取select查询的行数，该属性是为INSERT/UPDATE/DELETE这类DML操作设计的，大部分数据库驱动不会对SELECT查询返回正确的rowcount值。

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