如何在SQLAlchemy Session中设置执行选项并使用fetchmany？

你原写法失效的原因是：Session.execution_options()用于配置Session全局默认执行选项（比如autocommit、expire_on_commit），并非针对单次查询的链式调用。以下是两种可靠的实现方式：

方法1：单次查询直接传递执行选项 #

这是最简洁的方式，在execute()中直接指定stream_results=True：

from sqlalchemy import select

def process_large_table(db):
    # 执行查询时传入execution_options参数
    result = db.execute(
        select(EventConfigurationDB),
        execution_options={"stream_results": True}
    )
    # 批量获取数据，每次100条
    while batch := result.fetchmany(100):
        # 处理当前批次数据
        for row in batch:
            event_config = row[0]
            # 编写你的业务逻辑
            print(event_config.id)

方法2：复用底层连接的执行选项 #

如果需要对多个查询复用相同的执行配置，可以先获取Session绑定的数据库连接：

def process_large_table(db):
    # 获取Session底层连接并设置stream_results=True
    with db.connection().execution_options(stream_results=True) as conn:
        result = conn.execute(select(EventConfigurationDB))
        while batch := result.fetchmany(100):
            # 处理批次数据
            for row in batch:
                event_config = row[0]
                # 编写你的业务逻辑

额外注意 #

• 使用stream_results=True时，要确保处理完所有数据前Session/连接不会被关闭（比如FastAPI中，批量逻辑要在get_db()生成器的生命周期内完成）。
• fetchmany(size)在无数据时会返回空列表，因此可以通过while循环遍历所有批次。

内容的提问来源于stack exchange，提问作者Antonio Gamiz Delgado