使用pypyodbc从Azure Function连接Azure SQL的正确语法求助

你遇到的错误是因为pypyodbc不支持ActiveDirectoryManagedIdentity这个连接字符串参数，该参数是ODBC Driver 17+针对Azure AD托管身份的新认证模式，但pypyodbc的底层实现未适配该特性。以下是两种可行的解决方案：

方案一：改用pyodbc（推荐，兼容性更好） #

pyodbc对Azure SQL的Azure AD认证支持更完善，直接使用ActiveDirectoryManagedIdentity参数即可正常连接：

import pyodbc
import os

def main(req):
    # 从环境变量读取配置（避免硬编码）
    server = os.environ["SQL_SERVER"]
    database = os.environ["SQL_DATABASE"]
    
    # 构建连接字符串
    connection_string = f'Driver={{ODBC Driver 17 for SQL Server}};Server=tcp:{server}.database.windows.net,1433;Database={database};Authentication=ActiveDirectoryManagedIdentity;'
    
    try:
        conn = pyodbc.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM <table_name>')
        rows = cursor.fetchall()
        return str(rows)
    finally:
        # 确保连接关闭
        if conn:
            conn.close()

注意事项： #

• 确保Azure Function环境已安装pyodbc包（可在requirements.txt中添加pyodbc>=4.0.39）
• 为Function的托管身份（系统分配/用户分配）授予Azure SQL数据库的访问权限（如db_datareader角色）
• 环境中需安装ODBC Driver 17 for SQL Server（Azure Function的Python运行时默认已包含）

方案二：继续使用pypyodbc，通过MSI令牌连接 #

如果必须使用pypyodbc，可通过获取托管身份的访问令牌，以SQL认证模式连接：

import pypyodbc
import os
import requests

def get_msi_token():
    # 从环境变量获取MSI端点和密钥
    msi_endpoint = os.environ["MSI_ENDPOINT"]
    msi_secret = os.environ["MSI_SECRET"]
    # Azure SQL的资源URI
    resource = "https://database.windows.net/"
    
    headers = {
        "Secret": msi_secret,
        "Resource": resource
    }
    
    response = requests.get(msi_endpoint, headers=headers)
    response.raise_for_status()
    return response.json()["access_token"]

def main(req):
    server = os.environ["SQL_SERVER"]
    database = os.environ["SQL_DATABASE"]
    token = get_msi_token()
    
    # 用令牌作为密码，UID可填任意字符串
    connection_string = f'Driver={{ODBC Driver 17 for SQL Server}};Server=tcp:{server}.database.windows.net,1433;Database={database};UID=MSI_User;PWD={token};'
    
    try:
        conn = pypyodbc.connect(connection_string)
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM <table_name>')
        rows = cursor.fetchall()
        return str(rows)
    finally:
        if conn:
            conn.close()

注意事项： #

• 需为Function启用托管身份（系统分配或用户分配），并授予Azure SQL的访问权限
• UID字段可填写任意非空字符串，不影响身份验证
• 确保环境已安装requests包（requirements.txt中添加requests>=2.31.0）

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