使用Pandas读取存储过程报错：仅支持SQLAlchemy可连接对象

1. 连接忙碌错误：你在循环中复用了同一个pyodbc连接cnxn，该连接正被cursor.execute(storedProc)的结果集占用（遍历cursor时结果集未完全读取/释放），此时调用data.to_sql使用同一个连接执行新命令，导致连接冲突。
2. Pandas警告：data.to_sql传入的是pyodbc连接对象，而非SQLAlchemy的可连接对象（引擎/连接），Pandas对原生DBAPI连接的支持有限，因此触发警告。

统一使用SQLAlchemy引擎管理所有数据库操作，避免混合使用原生pyodbc连接，同时优化资源使用（不在循环内重复创建引擎）：

修改后的代码 #

import pyodbc
import sys
import os
from mymodules import ODBC, SendEmail, MyLogging
import pandas as pd
import gc
import sqlalchemy as sqlEngine
from sqlalchemy import text
import urllib

# 仅创建一次SQLAlchemy引擎，避免循环内重复创建
cnxn_str = "driver=ODBC+Driver+17+for+SQL+Server; server=server; database=db; uid=uid; pwd=pwd"
params = urllib.parse.quote_plus(cnxn_str)
engine = sqlEngine.create_engine(f'mssql+pyodbc:///?odbc_connect={params}')

# 执行截断表存储过程
with engine.connect() as conn:
    conn.execute(text("exec [dbo].[TruncateStageTablesETL]"))
    conn.commit()

# 获取GetMaxValuesForETL的结果，用SQLAlchemy读取避免连接占用
max_values_df = pd.read_sql_query("exec GetMaxValuesForETL", engine)

for _, row in max_values_df.iterrows():
    strObj = row[0]
    strTbl = row[2]
    param_val = row[1]
    
    # 使用参数化调用存储过程，避免SQL注入
    storedProc2 = text(f"exec :proc_name :param")
    data = pd.read_sql_query(storedProc2, engine, params={"proc_name": strObj, "param": param_val})
    
    # to_sql使用SQLAlchemy引擎，避免连接冲突
    data.to_sql(
        name=strTbl,
        schema='stage',
        con=engine,
        if_exists='append',
        index=False
    )
    
    del data
    gc.collect()

关键修改点 #

• 统一用SQLAlchemy引擎：所有数据库操作（执行存储过程、读取数据、写入数据）都通过SQLAlchemy引擎完成，由SQLAlchemy负责连接池管理，避免连接忙碌问题。
• 避免循环内重复创建引擎：引擎只初始化一次，减少资源开销。
• 参数化存储过程调用：使用text对象和params参数传递存储过程名和参数，避免字符串拼接带来的SQL注入风险。
• 用DataFrame遍历结果：通过pd.read_sql_query获取GetMaxValuesForETL的结果，转为DataFrame后遍历，不会占用连接导致后续操作冲突。

额外说明 #

如果你的ODBC.RunSQL方法是自定义的执行SQL函数，建议也改为使用SQLAlchemy引擎实现，保持数据库操作的一致性。

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