Sample to Call Azure API Management Operation with Service Principal

sample_to_call_azure_api_management_operation_with_service_principal.py

"""
This is a sample to call API management operation with OAuth based on Azure AD.
"""

import msal
import logging
import requests


if __name__ == "__main__":
    # application configuration
    config = {
        # Tenant ID of the application which receives the request
        "authority": "https://login.microsoftonline.com/55555555-6666-7777-8888-999999999999",
        # Client application ID of the application which sends the request
        "client_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
        # The scope of the application which sends the request
        "scope": ["api://00000000-1111-2222-3333-444444444444/.default"],
        # The private key file of the application which sends the request
        # The key must be used with the matching thumbprint.
        "thumbprint": "0123456789ABCDEF01234567890ABCDEF1234567",
        "private_key_file": "/sample/key.key",
    }

    # Create a preferably long-lived app instance that maintains a token cache.
    app = msal.ConfidentialClientApplication(
        config["client_id"], authority=config["authority"],
        client_credential={"thumbprint": config["thumbprint"], "private_key": open(config['private_key_file']).read()},
    )

    # The pattern to acquire a token looks like this.
    result = None

    # First, the code looks up a token from the cache.
    # Because we're looking for a token for the current app, not for a user,
    # use None for the account parameter.
    result = app.acquire_token_silent(config["scope"], account=None)

    if not result:
        logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
        result = app.acquire_token_for_client(scopes=config["scope"])

    if "access_token" in result:
        # Call a protected API with the access token.
        print(result["token_type"])
    else:
        print(result.get("error"))
        print(result.get("error_description"))
        print(result.get("correlation_id"))

    endpoint = 'https://test.sample.com/api/endpoint'
    http_headers = {'key_1': 'value_1',
                    'key_2': 'value_2'}
    data = requests.get(endpoint, headers=http_headers, stream=True)

    print(data.text)