Azure Functions部署报错：'Engine'对象无'execute'属性求解答

本地运行正常的代码部署到Azure Functions后报错，错误信息：

AttributeError: 'Engine' object has no attribute 'execute'

原代码 #

from sqlalchemy import create_engine, event
import textwrap
import base64
import pandas as pd
import logging
import pyodbc

def sql_run_query(data_base,service_date):
    driver = '{ODBC Driver 17 for SQL Server}'
    server_name = **********************************
    database_name = data_base
    server='{server_name},1433'.format(server_name=server_name)
    username = **********************************
    password = **********************************
    connection_string=textwrap.dedent('''Driver={driver};Server={server};Database={database};Uid={username};Pwd={password};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;'''.format(driver=driver,server=server,database=database_name,username=username,password=password))
    engine = create_engine('mssql+pyodbc:///?odbc_connect='+connection_string)
    #engine =create_sql_engine(data_base)
    engine.execute('declare @service_date_var as varchar(100) = '+service_date+'delete from [dbo].[performance] where service_date = @service_date_var')
    logging.info(f"query in {data_base} run succesfully")
    print(f"query in {data_base} run succesfully")

报错堆栈 #

Result: Failure Exception: AttributeError: 'Engine' object has no attribute 'execute'
Stack: File "/azure-functions-host/workers/python/3.9/LINUX/X64/azure_functions_worker/dispatcher.py", line 452, in _handle__invocation_request call_result = await self._loop.run_in_executor(
File "/usr/local/lib/python3.9/concurrent/futures/thread.py", line 58, in run result = self.fn(*self.args, **self.kwargs)
File "/azure-functions-host/workers/python/3.9/LINUX/X64/azure_functions_worker/dispatcher.py", line 718, in _run_sync_func return ExtensionManager.get_sync_invocation_wrapper(context,
File "/azure-functions-host/workers/python/3.9/LINUX/X64/azure_functions_worker/extension.py", line 215, in _raw_invocation_wrapper result = function(**args)
File "/home/site/wwwroot/BlobTrigger2/init.py", line 40, in main sql_run_query('kad_smc',dte)
File "/home/site/wwwroot/sql_uploader/sql_uploader.py", line 74, in sql_run_query engine.execute('''

错误原因 #

• 环境依赖版本不一致：本地使用SQLAlchemy 1.x，Azure环境自动安装了SQLAlchemy 2.x。SQLAlchemy 2.x移除了Engine.execute()方法，这是报错的核心原因。
• SQLAlchemy 2.x规范中，执行SQL必须通过Connection或Session对象，不再支持直接调用Engine的execute方法。

修复方案 #

方案1：适配SQLAlchemy 2.x（推荐） #

使用engine.connect()获取连接对象，配合参数化查询执行SQL，同时修复原代码的SQL注入风险：

from sqlalchemy import create_engine, text
import textwrap
import base64
import pandas as pd
import logging
import pyodbc

def sql_run_query(data_base,service_date):
    driver = '{ODBC Driver 17 for SQL Server}'
    server_name = **********************************
    database_name = data_base
    server='{server_name},1433'.format(server_name=server_name)
    username = **********************************
    password = **********************************
    connection_string=textwrap.dedent('''Driver={driver};Server={server};Database={database};Uid={username};Pwd={password};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;'''.format(driver=driver,server=server,database=database_name,username=username,password=password))
    engine = create_engine('mssql+pyodbc:///?odbc_connect='+connection_string)
    
    # 使用text()包裹SQL，参数化传递变量避免SQL注入
    delete_sql = text("DELETE FROM [dbo].[performance] WHERE service_date = :service_date")
    with engine.connect() as conn:
        conn.execute(delete_sql, {"service_date": service_date})
        conn.commit()  # 数据修改操作需要提交事务
    
    logging.info(f"Query in {data_base} run successfully")
    print(f"Query in {data_base} run successfully")

方案2：强制使用SQLAlchemy 1.x（临时兼容） #

在项目的requirements.txt中指定SQLAlchemy版本为1.x，例如：

SQLAlchemy==1.4.49
pyodbc==4.0.39
# 其他依赖按实际需求添加

重新部署Azure Functions，确保环境安装指定版本的依赖。

额外注意事项 #

• 原代码存在严重SQL注入风险：直接拼接用户输入到SQL语句中，必须通过参数化查询修复（方案1已处理）。
• 部署Azure Functions时，务必通过requirements.txt明确所有依赖的版本，避免本地与云端环境依赖不一致。

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