azure-sql-databaseflask-sqlalchemyazure-managed-identitypython-3.10

Use User Managed Identity (UMI) to authenticate Azure App Service (Python) to SQL Database


I am trying to connect to Azure SQL Database through my Flask application. I have created user managed identity in azure and assigned it to the app service, azure sql server, sql database. Even user is created for that UMI and has reader and writer access to database.

This is my test 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}')

how ever I am getting below error

"SQL Connection Unsuccesful. Error: (pyodbc.Error) ('FA001', '[FA001] [Microsoft][ODBC Driver 18 for SQL Server]Cannot use Authentication option with Integrated Security option. (0) (SQLDriverConnect)')\n(Background on this error at: http://sqlalche.me/e/14/dbapi)"

It does not connect with Trusted_Connection=false or Integrated_Security=false options added to connection string.


Solution

  • Bhavani's answer had almost everything correct, but one thing was missing. When we deploy this to an azure app based on Linux and app started using a Gunicorn command.

    You got to create the Service Connector in azure in order to get this working. We got this in an working session with Microsoft when we raised a ticket for the same.

    My working version:

    def mdbTest():
    try:
        managedIdentityId = request.args.get('managedIdentityId', '')
        server = ConfigSettings.SQLHost
        port = 1433
        database = ConfigSettings.SQLDatabase
        authentication = 'ActiveDirectoryMsi'  # The value should be 'ActiveDirectoryMsi'
    
        # For user-assigned managed identity.
        client_id = os.getenv('AZURE_CLIENT_ID') or managedIdentityId
        connString = f'Driver={{ODBC Driver 18 for SQL Server}};Server={server},{port};Database={database};UID={client_id};Authentication={authentication};Encrypt=yes;'
    
        conn = pyodbc.connect(connString)
        cursor = conn.cursor()
    
        # Execute a SQL query
        query = "SELECT top 10 percent * FROM dbo.tableName"  # Replace with your actual table name
        cursor. Execute(query)
    
        # Fetch all records
        records = cursor.fetchall()
        logger.error(f"SQL Connection Succesful. Activity Records: {len(records)}, Conn String: {connString}")
        return jsonify(f'SQL Connection Succesful. Activity Records: {len(records)}, Conn String: {connString}')
    except Exception as e:
        logger.error(f"SQL Connection Unsuccesful. Conn String: {connString} ,Error: {e}")
        return jsonify(f'SQL Connection Unsuccesful. Conn String: {connString} ,Error: {e}')