如何在SQLAlchemy中批量执行查询以减少数据库往返次数？

SQLAlchemy完全支持通过单次数据库往返获取多个独立查询的结果，核心依赖数据库对**多结果集（Multiple Result Sets）**的支持，主流数据库（如PostgreSQL、MySQL、SQLite 3.33+等）均提供该能力。

方法1：ORM原生多结果集查询（SQLAlchemy 2.0+推荐） #

利用session.execute同时执行多个ORM查询语句，通过Result对象的next()方法切换结果集，全程无需编写原生SQL：

from sqlalchemy import select
from sqlalchemy.orm import Session

def get_all_users_and_items(session: Session):
    # 构造两个独立的ORM查询
    user_stmt = select(User)
    item_stmt = select(Item)
    
    # 单次执行多个语句，返回多结果集
    result = session.execute(user_stmt, item_stmt)
    
    # 提取第一个结果集（User数据）
    users = result.scalars().all()
    # 切换到下一个结果集
    result = result.next()
    # 提取第二个结果集（Item数据）
    items = result.scalars().all()
    
    return users, items

方法2：组合SQL语句执行（兼容旧版本SQLAlchemy） #

如果使用SQLAlchemy 1.x版本，可通过组合原生SQL语句的方式，一次性执行两个查询，再手动将结果映射到ORM模型：

from sqlalchemy import text
from sqlalchemy.orm import Session

def get_all_users_and_items(session: Session):
    # 组合两个SELECT语句（数据库需支持多语句执行）
    combined_sql = text("SELECT id, name FROM users; SELECT id, name, desc FROM items;")
    result_proxy = session.execute(combined_sql)
    
    # 提取第一个结果集并映射为User对象
    users = [User(id=row.id, name=row.name) for row in result_proxy.fetchall()]
    # 切换到下一个结果集
    result_proxy.nextset()
    # 提取第二个结果集并映射为Item对象
    items = [Item(id=row.id, name=row.name, desc=row.desc) for row in result_proxy.fetchall()]
    
    return users, items

关键说明 #

• 两种方法均只触发一次数据库往返，所有数据通过单次请求从数据库获取，对比原代码的两次请求，能有效减少网络开销。
• 需确保使用的数据库支持多结果集/多语句执行，主流数据库均兼容，但部分嵌入式或小众数据库可能不支持。
• 方法1更贴合ORM的使用习惯，无需手动维护原生SQL与模型的映射关系，推荐在SQLAlchemy 2.0及以上版本优先使用。

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