DeepankarB
New Contributor III

I have been able to resolve this issue. Apparently you need to generate access token using service principal client id and client secret.  

saurabh18cs solution is more relevant to Azure Databricks. Got below link from Databricks which provide generic solution for using service principal: https://docs.databricks.com/en/dev-tools/auth/oauth-m2m.html#manually-generate-a-workspace-level-acc... 
 
 
 
 
Here is sample code for reference. Ensure that service principal has access to relevant services.
 

 

import os
import json
import requests
from requests.auth import HTTPBasicAuth


# Set environment variables
os.environ["DATABRICKS_INSTANCE"] = "https://your-databricks-instance"
os.environ["DATABRICKS_TOKEN"] = "your-personal-access-token"
os.environ["DATABRICKS_CLIENT_ID"] = "your-client-id"
os.environ["DATABRICKS_CLIENT_SECRET"] = "your-client-secret"
os.environ["SPFLAG"] = "True"



def get_env_var(var_name):
    """Fetch environment variable with error handling."""
    value = os.getenv(var_name)
    if not value:
        raise ValueError(f"{var_name} is not set.")
    return value


def get_access_token_via_sp(token_endpoint_url, client_id, client_secret):
    """Generate an access token using service principal credentials."""
    data = {
        'grant_type': 'client_credentials',
        'scope': 'all-apis'
    }
    response = requests.post(
        url=token_endpoint_url,
        auth=HTTPBasicAuth(client_id, client_secret),
        data=data
    )

    if response.status_code == 200:
        print("Successfully generated token.")
        return response.json().get("access_token")
    else:
        raise Exception(f"Error {response.status_code}: {response.text}")


def fetch_databricks_clusters(databricks_instance, access_token):
    """Fetch list of Databricks clusters."""
    url = f"{databricks_instance}/api/2.0/clusters/list"
    headers = {"Authorization": f"Bearer {access_token}"}

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        print("Cluster list fetched successfully.")
        return response.json()
    else:
        raise Exception(f"Error {response.status_code}: {response.text}")


def main():
    # Environment variables
    databricks_instance = get_env_var("DATABRICKS_INSTANCE")
    spflag = os.getenv("SPFLAG", "False").lower() == "true"

    # Access token retrieval
    if spflag:
        token_endpoint_url = f"{databricks_instance}/oidc/v1/token"
        client_id = get_env_var("DATABRICKS_CLIENT_ID")
        client_secret = get_env_var("DATABRICKS_CLIENT_SECRET")
        access_token = get_access_token_via_sp(token_endpoint_url, client_id, client_secret)
    else:
        access_token = get_env_var("DATABRICKS_TOKEN")

    print(f"Using Databricks token: {access_token}")

    # Fetch clusters
    clusters = fetch_databricks_clusters(databricks_instance, access_token)
    print(json.dumps(clusters, indent=2))


if __name__ == "__main__":
    main()

 

 
 
 
 
 

View solution in original post