Python使用PYODBC/SQLAlchemy获取长查询已处理行状态的方法咨询

核心结论 #

可以实现，分两种场景匹配你的需求：基于行读取计数的进度统计，以及基于SQL Server内置DMV的查询进度统计。 #

方案1：基于结果行读取计数（适配「已接收X行」展示场景） #

这个方案最贴近你观察到的应用进度展示效果，逻辑如下：

• PYODBC和SQLAlchemy都支持流式读取查询结果，不需要等服务端返回全部结果集才开始处理，你可以每读取一行/一批行就累加计数，实时更新已接收行数。
• PYODBC原生支持cursor.execute()后逐行fetch，SQLAlchemy可以通过设置stream_results=True开启流式读取，避免客户端一次性加载全部结果占用过多内存。

PYODBC 代码示例 #

import pyodbc
conn = pyodbc.connect("DRIVER={ODBC Driver 17 for SQL Server};SERVER=你的实例地址;DATABASE=目标库名;UID=登录账号;PWD=登录密码")
cursor = conn.cursor()
# 执行耗时查询
cursor.execute("你的耗时SELECT查询语句")
row_count = 0
while True:
    row = cursor.fetchone()
    if not row:
        break
    row_count += 1
    # 此处可将row_count输出到日志、前端等位置展示进度
    print(f"已接收{row_count}行")

SQLAlchemy 代码示例 #

from sqlalchemy import create_engine, text

engine = create_engine("mssql+pyodbc://账号:密码@实例地址/库名?driver=ODBC+Driver+17+for+SQL+Server")
with engine.connect().execution_options(stream_results=True) as conn:
    result = conn.execute(text("你的耗时SELECT查询语句"))
    row_count = 0
    for row in result:
        row_count +=1
        print(f"已接收{row_count}行")

方案2：基于线程轮询的全局查询进度统计 #

如果你的查询是无返回结果的写操作（比如大表UPDATE、DELETE、建索引），或者你需要获取SQL Server端的执行进度而非客户端接收行数，可以用该方案：

• 把耗时查询放到子线程执行，主线程定期轮询SQL Server的动态管理视图（DMV）sys.dm_exec_requests，该视图的percent_complete字段会返回以下操作的执行进度：
  • ALTER INDEX REORGANIZE
  • 数据库自动收缩
  • 数据库备份
  • 索引创建
  • DBCC CHECKDB、CHECKTABLE等完整性检查操作
  • 数据库收缩
  • 数据库恢复
  • 事务回滚
  • TDE加密扫描
• 注意普通SELECT查询不会填充percent_complete字段，这类场景优先使用方案1的行计数逻辑。

实现逻辑 #

1. 子线程执行耗时查询前，先获取当前连接的session_id，存入线程安全的共享变量
2. 主线程每隔1-5秒执行一次查询：SELECT percent_complete, total_elapsed_time FROM sys.dm_exec_requests WHERE session_id = 对应会话ID，拿到进度后更新展示
3. 子线程执行结束后终止轮询即可

注意事项 #

• 流式读取场景下不要提前调用fetchall()，会导致客户端一次性加载全部结果，无法逐行计数
• 跨线程共享session_id、行计数变量时需要加锁，避免并发读写冲突
• 如果查询包含大字段（TEXT、BLOB等类型），逐行fetch会有小幅性能损耗，可以调整为每次批量读取再更新计数，比如用fetchmany(1000)每次读取1000行后再更新进度，平衡性能和进度更新的实时性

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