topic Error API calling with Service Principal Secret in Data Engineering

Error API calling with Service Principal Secret

DeepankarB — Mon, 13 Jan 2025 09:03:49 GMT

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}")

Re: Error API calling with Service Principal Secret

saurabh18cs — Mon, 13 Jan 2025 09:56:30 GMT

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}")

Re: Error API calling with Service Principal Secret

DeepankarB — Tue, 14 Jan 2025 23:59:53 GMT

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-access-token

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