使用UMI认证Azure App Service(Python)连接SQL数据库遇错

问题原因 #

错误提示Cannot use Authentication option with Integrated Security option，核心原因是SQLAlchemy的mssql+pyodbc驱动默认会自动添加Integrated Security=SSPI参数，和你指定的Authentication选项冲突，即使手动添加Trusted_Connection=false也会因为参数优先级或格式问题不生效。

解决方案 #

1. 修正连接字符串构造 #

修改连接字符串，明确禁用Integrated Security，并使用适配ODBC Driver 18的Authentication值：

# 需先导入urllib.parse模块
import urllib.parse

# 替换原连接字符串构造逻辑
odbc_conn_str = f"Driver={driver};Server=tcp:{sqlHost},1433;Database={databaseName};UID={managedIdentityId};Authentication=ActiveDirectoryManagedIdentity;Integrated Security=False"
encoded_odbc_str = urllib.parse.quote_plus(odbc_conn_str)
sqlalchemy_connection_string = f'mssql+pyodbc:///?odbc_connect={encoded_odbc_str}'

关键说明：

• 用urllib.parse.quote_plus对ODBC参数做URL编码，避免特殊字符干扰连接
• 明确指定Integrated Security=False覆盖驱动默认值
• 替换ActiveDirectoryMsi为ActiveDirectoryManagedIdentity，适配ODBC Driver 18的最新规范
• 补充SQL Server端口1433，确保连接地址完整

2. 简化冗余配置（可选） #

如果你的Azure SQL实例使用默认信任证书，无需手动指定ssl_ca，可以直接移除connect_args参数，简化引擎初始化代码：

engine = create_engine(sqlalchemy_connection_string, echo=False)

若必须指定证书，需确保myCertificate.crt.pem文件路径正确，且App Service有权限访问该文件。

3. 验证UMI权限配置 #

再次确认以下配置是否正确：

• UMI已成功分配给目标App Service
• UMI在Azure SQL Server的IAM角色中拥有SQL Server Contributor或等效权限
• 已在SQL数据库中为UMI创建对应用户，并授予读写权限（示例SQL语句）：

CREATE USER [你的UMI名称] FROM EXTERNAL PROVIDER;
ALTER ROLE db_datareader ADD MEMBER [你的UMI名称];
ALTER ROLE db_datawriter ADD MEMBER [你的UMI名称];

修正后的完整代码片段 #

import urllib.parse
from flask import request, jsonify
from sqlalchemy import create_engine
import logging

logger = logging.getLogger(__name__)

@app.route('/dbconnectManaged/', methods=['GET'])
def dbconnectionManaged():
    try:
        sqlHost = request.args.get('sqlHost', '') or ConfigSettings.MSSQLHost
        databaseName = request.args.get('database', '') or ConfigSettings.MSSQLDatabase
        driver = 'ODBC Driver 18 for SQL Server'
        
        managedIdentityId = ConfigSettings.ManagedIdentityId
        if managedIdentityId in ['none', '', 'None', None]:
            logger.info('Managed Identity Id cannot be null or empty')
            return jsonify('Managed Identity Id cannot be null or empty')
        
        # 构造编码后的ODBC连接字符串
        odbc_conn_str = f"Driver={driver};Server=tcp:{sqlHost},1433;Database={databaseName};UID={managedIdentityId};Authentication=ActiveDirectoryManagedIdentity;Integrated Security=False"
        encoded_odbc_str = urllib.parse.quote_plus(odbc_conn_str)
        sqlalchemy_connection_string = f'mssql+pyodbc:///?odbc_connect={encoded_odbc_str}'

        logger.info(f'Conn Str: {sqlalchemy_connection_string}')
        engine = create_engine(sqlalchemy_connection_string, echo=False)
        dbConnection = engine.raw_connection()
        logger.info('Successfully Established SQL Connection')
        return jsonify('Successfully Established SQL Connection')
        
    except Exception as e:
        logger.error(f"SQL Connection Unsuccesful. Error: {e}")
        return jsonify(f'SQL Connection Unsuccesful. Error: {e}')

内容的提问来源于stack exchange，提问作者Mayur A Muley