cancel
Showing results for 
Search instead for 
Did you mean: 
Data Engineering
Join discussions on data engineering best practices, architectures, and optimization strategies within the Databricks Community. Exchange insights and solutions with fellow data engineers.
cancel
Showing results for 
Search instead for 
Did you mean: 

Error API calling with Service Principal Secret

DeepankarB
New Contributor III

Hi,

I am working on Databricks workspace setup on AWS and trying to use Service Principal to execute API calls (CI/CD) deployment through Bitbucket. So I created secret for the service principal and trying to test the token. The test failed with below error:

Using token: doseXXXXXXXXXXXXXXXXXXX calling request get Error 401: {"error_code":401,"message":"Credential was not sent or was of an unsupported type for this API."}

I changed the token to PAT and the script worked fine. Can you please confirm what the issue here. The service principal looks fine as I tested running DLT pipelines successfully via service principal.

Here is the script for your reference. This code list the clusters in the workspace:

 

 

 

import os
import requests

# Fetch the token from environment variables
databricks_token = 'doseXXXXXXXXXXXXXXXXXXX'

if not databricks_token:
    raise ValueError("DATABRICKS_TOKEN is not set.")

print(f"Using token: {databricks_token}")

# Define the API endpoint and headers
databricks_instance = "https://xxx.databricks.com"  # Replace with your Databricks workspace URL
url = f"{databricks_instance}/api/2.0/clusters/list"
headers = {"Authorization": f"Bearer {databricks_token}"}

# Make the API call
print("calling request get")
response = requests.get(url, headers=headers)

# Check response
if response.status_code == 200:
    print("Success:", response.json())
    # display(response.json())
else:
    print(f"Error {response.status_code}: {response.text}")

 

 

 

1 ACCEPTED SOLUTION

Accepted Solutions

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

2 REPLIES 2

saurabh18cs
Valued Contributor III

Hi , Use following oauth token instead. make changes for AWS

 
sample python code:
 

import requests
import json
import os

# Environment variables (replace with your actual values or use environment variables)
DATABRICKS_WORKSPACE_URL = os.getenv("DATABRICKS_WORKSPACE_URL")
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
TENANT_ID = "<<TENANTID>>"
SCOPE = "2ff814a6-3304-4ab8-85cb-cd0e6f879c1d/.default"

# URLs
TOKEN_URL = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
DATABRICKS_URL = f"{DATABRICKS_WORKSPACE_URL}/api/2.0/token/create"

# Get access token
payload = {
"client_id": CLIENT_ID,
"grant_type": "client_credentials",
"scope": SCOPE,
"client_secret": CLIENT_SECRET
}

response = requests.post(TOKEN_URL, data=payload)
response.raise_for_status() # Raise an error for bad status codes
access_token_val = response.json()
access_token = access_token_val.get("access_token")

if not access_token:
print("Failed to obtain access token")
exit(1)

print(f"Access Token: {access_token}")

# Create Databricks token
headers = {
"Authorization": f"Bearer {access_token}",
"X-Databricks-Azure-SP-Management-Token": access_token,
"Content-Type": "application/json"
}

data = {
"comment": "pipeline token"
}

api_response = requests.post(DATABRICKS_URL, headers=headers, json=data)
api_response.raise_for_status() # Raise an error for bad status codes
api_response_json = api_response.json()
DATABRICKS_NEW_TOKEN = api_response_json.get("token_value")

if not DATABRICKS_NEW_TOKEN:
print("Token could not be created")
exit(1)
else:
print("Successfully created a Databricks Token")
print(f"##vso[task.setvariable variable=DATABRICKS_TOKEN;isOutput=true]{DATABRICKS_NEW_TOKEN}")
print(f"##vso[task.setvariable variable=ACCESS_TOKEN;isOutput=true]{access_token}")

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()

 

 
 
 
 
 

Connect with Databricks Users in Your Area

Join a Regional User Group to connect with local Databricks users. Events will be happening in your city, and you won’t want to miss the chance to attend and share knowledge.

If there isn’t a group near you, start one and help create a community that brings people together.

Request a New Group