Python Azure函数中返回数据库超时自定义错误的实现方法

首先修正代码里的变量名错误，这些错误会直接导致未捕获的异常：

• connection()函数里返回的conctn需改为connection
• read_table函数里的conctn.timeout需改为cnxn.timeout

接下来，通过解析pyodbc.OperationalError的错误码，精准捕获超时场景并返回自定义信息：

1. 带超时控制的连接函数 #

import pyodbc
import pandas as pd

def connection():
    username = ''
    server = ''
    password = ''
    database = ''
    # 拼接连接字符串，添加连接超时参数
    conn_str = (
        'DRIVER={ODBC Driver 17 for SQL Server};'
        f'SERVER={server};'
        f'DATABASE={database};'
        'ENCRYPT=yes;'
        f'UID={username};'
        f'PWD={password};'
        'CONNECT_TIMEOUT=3'  # 设置连接超时时间（单位：秒）
    )
    try:
        connection = pyodbc.connect(conn_str)
        return connection
    except pyodbc.OperationalError as e:
        # 匹配连接超时相关错误码（08001/HYT00）
        if any(code in str(e) for code in ['08001', 'HYT00']):
            return {"error": "数据库连接超时，请稍后重试"}
        raise  # 非超时异常继续抛出，便于排查其他问题

2. 处理查询超时的读表函数 #

def read_table(table, cnxn):
    cnxn.timeout = 3  # 设置查询超时时间（单位：秒）
    query = f'SELECT * FROM {table}'
    try:
        df = pd.read_sql(query, cnxn)
    except pyodbc.OperationalError as e:
        # 匹配查询超时错误码HYT00
        if 'HYT00' in str(e):
            raise Exception("数据查询超时，请稍后重试")
        raise
    if df.empty:
        raise Exception("查询结果为空")
    return df

3. 主函数逻辑（Azure函数示例） #

def main(req):
    try:
        conn = connection()
        # 判断连接是否返回错误信息
        if isinstance(conn, dict):
            return conn
        df1 = read_table("table_name", conn)
        # 后续业务逻辑...
        return {"data": df1.to_dict('records')}
    except Exception as e:
        # 捕获所有自定义异常，返回统一格式的错误信息
        return {"error": str(e)}

关键说明 #

• 错误码匹配：HYT00是ODBC标准的查询超时错误码，08001对应连接超时类错误，通过解析异常字符串精准识别超时场景。
• 避免默认日志报错：所有超时异常都被捕获并转换为自定义响应，不会让原始pyodbc异常流入Azure函数的默认日志系统。
• 安全提示：当前SELECT * FROM {table}存在SQL注入风险，建议改用参数化查询：

  query = 'SELECT * FROM ?'
  df = pd.read_sql(query, cnxn, params=(table,))
  

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