Hello
I have written a python script that uses Databricks Rest API(s). I am trying to clone/ update an Azure Devops Repository inside databricks using Azure Service Principal. I am able to retrieve the credential_id for the service principal I am using. Every time I try to clone or update the repo I get authentication error. I am unable to understand how to use it for authentication for cloning of the repo. Here's the script:
import getopt
import json
import sys
import requests
def create_git_credentials(host_name, token, devops_pat):
method = "GET"
url = f"{host_name}api/2.0/git-credentials"
headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
payload = json.dumps({
"git_username": "testuser",
"git_provider": "azureDevOpsServices",
"personal_access_token": devops_pat
})
response = requests.request(method, url, headers=headers, data=payload)
print('RESPONSE:', response.json())
if response.status_code == 200:
credentials_id = response.json()['credentials'][0]['credential_id']
return credentials_id
else:
raise ValueError(f"Error creating git credentials: {response.text}")
def checkIfRepoExists(host_name, token, repo_path_in_dbks):
method = "GET"
url = f"{host_name}api/2.0/repos"
headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
response = requests.request(method, url, headers=headers)
response_json = json.loads(response.text)
listToFilter = list(response_json["repos"])
while len(response_json) > 1:
new_response = requests.request(
method,
url,
headers=headers,
params={"next_page_token": response_json["next_page_token"]},
)
response_json = json.loads(new_response.text)
listToFilter.extend(response_json["repos"])
filtered_list = [
dictionary
for dictionary in listToFilter
if repo_path_in_dbks == dictionary["path"]
]
if len(filtered_list) == 0:
return 0
repo_id = filtered_list[0]["id"]
return repo_id
def createParentDirectoryIfNotExists(host_name, token, repo_path_in_dbks):
parent_path = "/".join(repo_path_in_dbks.split("/")[:-1])
method = "GET"
url = f"{host_name}api/2.0/workspace/list"
headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
response = requests.request(method, url, headers=headers, params={"path": parent_path})
response_json = json.loads(response.text)
if 'error_code' in response_json and response_json['error_code'] == 'RESOURCE_DOES_NOT_EXIST':
method = "POST"
url = f"{host_name}api/2.0/workspace/mkdirs"
payload = json.dumps({"path": parent_path})
response = requests.request(method, url, headers=headers, data=payload)
if response.status_code != 200:
raise ValueError(f"Failed to create parent directory {parent_path}: {response.text}")
def updateRepo(host_name, token, repoId, branch):
method = "PATCH"
url = f"{host_name}api/2.0/repos/{str(repoId)}"
headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
payload = json.dumps({"branch": branch})
print("payload is\n" + payload)
print(method + "-Request: " + url + "\n" + "payload:\n" + payload)
response = requests.request(method, url, headers=headers, data=payload)
if response.status_code == 200:
return response.text
else:
raise ValueError(response.text)
def cloneRepo(host_name, token, git_url, repo_path_in_dbks, branch):
method = "POST"
url = f"{host_name}api/2.0/repos"
headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
payload = json.dumps({
"url": git_url,
"provider": "azureDevOpsServices",
"path": repo_path_in_dbks,
"branch": branch
})
print("payload is\n" + payload)
print(method + "-Request: " + url + "\n" + "payload:\n" + payload)
response = requests.request(method, url, headers=headers, data=payload)
print("RESPONSE:", response.json())
if response.status_code == 200:
return response.text
else:
raise ValueError(response.text)
def main():
try:
opts, args = getopt.getopt(
sys.argv[1:],
"hstcld",
["host_name=", "token=", "git_url=", "branch=", "repo_path_in_dbks=", "devops_pat="],
)
except getopt.GetoptError:
print(
"create_update_dbks_repo.py"
+ " -h <host_name>"
+ " -t <token>"
+ " -g <git_url>"
+ " -b <branch>"
+ " -p <repo_path_in_dbks>"
+ " -d <devops_pat>"
)
sys.exit(2)
for opt, arg in opts:
if opt in ("-h", "--host_name"):
host_name = arg
elif opt in ("-t", "--token"):
token = arg
elif opt in ("-g", "--git_url"):
git_url = arg
elif opt in ("-b", "--branch"):
branch = arg
elif opt in ("-p", "--repo_path_in_dbks"):
repo_path_in_dbks = arg
elif opt in ("-d", "--devops_pat"):
devops_pat = arg
print("-h is " + host_name)
print("-g is " + git_url)
print("-b is " + branch)
print("-p is " + repo_path_in_dbks)
print("-d is " + devops_pat)
credentials_id = create_git_credentials(host_name, token, devops_pat)
repoName = git_url.split("/")[-1]
repoId = checkIfRepoExists(host_name, token, repo_path_in_dbks)
if repoId != 0:
updateRepo(host_name, token, repoId, branch)
print("DBKS Repo: " + repoName + " updated successfully.")
else:
createParentDirectoryIfNotExists(host_name, token, repo_path_in_dbks)
cloneRepo(host_name, token, git_url, repo_path_in_dbks, branch)
print("DBKS Repo: " + repoName + " cloned successfully.")
if __name__ == "__main__":
main()
Is there a way to directly use the DevOps PAT to authenticate with hitting the get_credentials endpoint?
