使用用户管理身份 (UMI) 对 Azure 应用服务 (Python) 进行 SQL 数据库身份验证

Question

我正在尝试通过 Flask 应用程序连接到 Azure SQL 数据库。我已经在 azure 中创建了用户管理身份，并将其分配给应用程序服务、azure sql 服务器、sql 数据库。甚至用户也是为该 UMI 创建的，并且具有数据库的读取者和写入者访问权限。

这是我的测试 API -

@app.route('/dbconnectManaged/', methods=['GET'])
def dbconnectionManaged():
    try:
        managedIdentityId = request.args.get('managedIdentityId', '')
        authentication = request.args.get('authentication', 'ActiveDirectoryMsi')
        sqlHost = request.args.get('sqlHost', '') or ConfigSettings.MSSQLHost
        databaseName = request.args.get('database', '') or ConfigSettings.MSSQLDatabase
        connect_args = {'ssl_ca': 'myCertificate.crt.pem'}
        conn_str = ''
        driver = 'ODBC Driver 18 for SQL Server'
        
        if ConfigSettings.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')
        else :
            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')
        else:
            sqlalchemy_connection_string = f'mssql+pyodbc://{sqlHost}/{databaseName}?driver={driver}&UID={managedIdentityId}&Authentication={authentication}'

        if sqlalchemy_connection_string != '':
            logger.info(f'Conn Str: {sqlalchemy_connection_string}')
            engine = create_engine(sqlalchemy_connection_string, connect_args=connect_args, echo=False)
            dbConnection = engine.raw_connection()
            logger.info('Successfully Established SQL Connection')
            return jsonify('Successfully Established SQL Connection')
        else:
            logger.error("Database Configuration Is Not Defined")
            return jsonify("No database connection defined. Please define database or contact support for assistance.")
    except Exception as e:
        logger.error(f"SQL Connection Unsuccesful. Error: {e}")
        return jsonify(f'SQL Connection Unsuccesful. Error: {e}')

我怎么会出现以下错误

“SQL 连接失败。错误：(pyodbc.Error) ('FA001'、'[FA001] [Microsoft][ODBC Driver 18 for SQL Server]无法将身份验证选项与集成安全选项一起使用。(0) (SQLDriverConnect)') （此错误的背景位于：http://sqlalche.me/e/14/dbapi）”

它不会与添加到连接字符串的 Trusted_Connection=false 或 Integrated_Security=false 选项进行连接。

Answer 1

sqlalchemy_connection_string = f'mssql+pyodbc://{sqlHost}/{databaseName}?driver={driver}&UID={managedIdentityId}&Authentication={authentication}'

您在连接字符串 (

UID={managedIdentityId}

) 中指定 UID（用户 ID）以及身份验证，这就是应用程序采用集成安全选项作为身份验证选项的原因。

“SQL 连接失败。错误：(pyodbc.Error) ('FA001'、'[FA001] [Microsoft][ODBC Driver 18 for SQL Server]无法将身份验证选项与集成安全选项一起使用。(0) (SQLDriverConnect)') （此错误的背景位于：http://sqlalche.me/e/14/dbapi）”

这些选项是冲突的，因为 MSI 身份验证应该在内部处理身份验证，而不需要单独指定 UID。这可能是出现上述错误的原因。要解决此问题，您应该调整 SQL Alchemy 连接字符串以正确利用 MSI 身份验证，而无需指定 UID。以下是修改连接字符串的方法：

sqlalchemy_connection_string = f'mssql+pyodbc://{sqlHost}/{databaseName}?driver={driver}&Authentication={authentication}'

然后您就可以成功连接SQL数据库，不会出现任何错误。或者根据MS socument，您也可以使用以下连接字符串格式：

{ODBC Driver 18 for SQL Server};Server=tcp:<database-server-name>.database.windows.net,1433;Database=<database-name>;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30

您可以使用以下代码：

import urllib.parse
from sqlalchemy import create_engine

server = '<serverName>.database.windows.net'
database = '<datbaseName>'

driver = 'ODBC Driver 17 for SQL Server'
conn = f'DRIVER={driver};SERVER={server};DATABASE={database};Authentication=ActiveDirectoryMsi'
conn_str = urllib.parse.quote_plus(conn)

engine = create_engine(f'mssql+pyodbc:///?odbc_connect={conn_str}', fast_executemany=True)

with engine.connect() as connection:
    result = connection.execute("select * from [dbo].[ticker]")
    for row in result:
        print(row)

它将成功连接到Azure SQL数据库，没有任何错误。