使用SQLAlchemy与Pandas查询SQL Server时如何设置查询超时？

你的代码未配置查询超时参数，所以15秒的SQL语句会完整执行。以下是两种有效的超时设置方法：

方法一：创建引擎时通过connect_args指定查询超时 #

直接在create_engine中传递query_timeout参数（单位：秒），该参数会传递给底层pyodbc连接：

import datetime
import pandas as pd
import sqlalchemy

connection_str = "mssql+pyodbc://@ITDEV/FACETS?trusted_connection=yes&driver=ODBC+Driver+17+for+SQL+Server"
engine = sqlalchemy.create_engine(
    connection_str,
    connect_args={"query_timeout": 3}  # 设置3秒查询超时
)

with engine.begin() as conn:
    print(datetime.datetime.now())
    try:
        pd.read_sql("WAITFOR DELAY '00:00:15'; SELECT TOP 1 * FROM CMC_SBSB_SUBSC", conn)
    except Exception as e:
        print(f"触发超时或其他错误：{e}")
    print(datetime.datetime.now())

方法二：在连接内部设置底层pyodbc连接的超时属性 #

通过SQLAlchemy连接对象获取底层pyodbc连接，直接设置其timeout属性：

import datetime
import pandas as pd
import sqlalchemy

connection_str = "mssql+pyodbc://@ITDEV/FACETS?trusted_connection=yes&driver=ODBC+Driver+17+for+SQL+Server"
engine = sqlalchemy.create_engine(connection_str)

with engine.begin() as conn:
    # 获取底层pyodbc连接并设置3秒超时
    conn.connection.timeout = 3
    print(datetime.datetime.now())
    try:
        pd.read_sql("WAITFOR DELAY '00:00:15'; SELECT TOP 1 * FROM CMC_SBSB_SUBSC", conn)
    except Exception as e:
        print(f"触发超时或其他错误：{e}")
    print(datetime.datetime.now())

注意事项 #

• 两种方法的超时单位均为秒，设置后超过指定时间未完成的查询会抛出pyodbc.OperationalError异常。
• 需确保使用的ODBC Driver 17 for SQL Server版本支持该参数，该版本原生支持查询超时配置。
• 必须添加异常捕获逻辑，避免超时直接导致程序崩溃。

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