pythonazure-devopsazure-devops-rest-apiserviceconnection
How to update an Azure Service Endpoint using python?
Currently I have a python function that takes the information of an Azure DevOps Service Connection. The problem is that I want to update that, and using a PUT request I always have a body problem. I´ve tried to use Postman to test if it works, and it is. The problem only remains on the python code. My code is this:
def update_sc_key(secret_value, app_name, sc_dependence):
print(f"SC DEPENDENCE:",sc_dependence)
personal_access_token = get_pat()
bearer = get_bearer()
if sc_dependence == True:
pat_credentials = BasicAuthentication('', personal_access_token)
connection = Connection(base_url=organization_url, creds=pat_credentials)
service_connection_name = app_name[len('SP_') + app_name[len('SP_'):].find('_') - 4:]
print(service_connection_name)
service_endpoint_client = connection.clients.get_service_endpoint_client()
service_connections = service_endpoint_client.get_service_endpoints(project_name)
for service_connection in service_connections:
if service_connection.name == "RENEWAL_SC_TEST": # service_connection_name
print(f'SERVICE CONNECTION NAME:', service_connection.name)
service_connection_id = service_connection.id
print(f'SERVICE CONNECTION ID:', service_connection_id)
# Get Service Connection body
project_url = organization_url + project_name + "/_apis/serviceendpoint/endpoints"
try:
# GET RESPONSE
get_params = {
"endpointIds": service_connection_id,
"api-version": "7.1-preview.4"
}
get_headers={
"Authorization": "Bearer " + bearer
}
get_response = requests.get(project_url, params=get_params, headers=get_headers)
get_response_tr = get_response.json()
get_data_obj1 = get_response_tr['value']
get_data_obj2 = str(get_data_obj1)
get_data_obj3 = get_data_obj2[1:-1]
if get_response.status_code == 200:
get_data_obj_mod = get_data_obj3.replace("'authenticationType': 'spnKey'", "'authenticationType': 'spnKey', 'serviceprincipalkey': '" + secret_value + "'")
get_data_obj_mod_rep = get_data_obj_mod.replace("'", "\"").replace("True", "true").replace("False", "false")
get_data_obj_mod_rep = json.load(get_data_obj_mod)
print(get_data_obj_mod_rep)
else:
print(f"Error: {get_response.status_code} - {get_response.text}")
# PUT RESPONSE TO CHANGE PWD
put_params = {
"endpointId": service_connection_id,
"api-version": "7.1-preview.4"
}
put_headers={
"Authorization": "Bearer " + bearer,
"Content-Type": "application/json"
}
put_body={
"body": get_data_obj_mod_rep
}
put_response = requests.put(project_url, params=put_params, headers=put_headers, data=put_body)
print(put_response.status_code)
print(put_response.text)
except json.JSONDecodeError as e:
print("ERROR: expired bearer token")
else:
print(f'No Service Connection')
The result of the variable "get_data_obj_mod_rep" works on Postman as I said, but if I use the same on python, I get this error:
{"$id":"1","innerException":null,"message":"Value cannot be null.\r\nParameter name: endpoint","typeName":"System.ArgumentNullException, mscorlib","typeKey":"ArgumentNullException","errorCode":0,"eventId":0}
Solution
Based on your python sample, I can reproduce the similar issue.
To solve this issue, I change to use another python format to get the service connection body and modify it.
Here is an example:
import requests
import json
url = "https://dev.azure.com/org/projectname/_apis/serviceendpoint/endpoints/serviceconnectionid?api-version=7.1-preview.4"
headers = {
'Content-Type': 'application/json',
'Authorization': 'Basic Based64token',
}
get_response = requests.request("Get", url, headers=headers)
secret_value = "xxxxx"
get_data_obj_mod = json.loads(get_response.content)
get_data_obj_mod["authorization"]["parameters"]["serviceprincipalkey"] = secret_value
ref_format = json.dumps(get_data_obj_mod, indent=2)
response = requests.request("PUT", url, headers=headers, data=ref_format)
print(response.text)
In this case, I can update the service connection successfully.
Update:
import requests
import json
url = "https://dev.azure.com/org/project/_apis/serviceendpoint/endpoints/serviceconnectionid?api-version=7.1-preview.4"
get_headers = {
'Content-Type': 'application/json',
'Authorization': 'Basic based64PAT',
}
get_response = requests.get(url, headers=get_headers)
secret_value = "test"
get_data_obj_mod = json.loads(get_response.content)
get_data_obj_mod["authorization"]["parameters"]["serviceprincipalkey"] = secret_value
ref_format = json.dumps(get_data_obj_mod, indent=2)
put_headers = {
'Content-Type': 'application/json',
'Authorization': 'Basic Based64PAT',
}
put_response = requests.put(url, headers=put_headers, data=ref_format)
print(put_response.text)
Update1:
When we use the params as the sample you shared, the url will change to below format:
Put https://dev.azure.com/org/project/_apis/serviceendpoint/endpoints/?endpointIds=serviceconnectionid&api-version=7.1-preview.4
This is not correct.
The correct format:
Put https://dev.azure.com/org/projectname/_apis/serviceendpoint/endpoints/serviceconnectionid?api-version=7.1-preview.4
This should be the root cause of the issue.
The following format will work too:
import requests
import json
project_url = "https://dev.azure.com/org/project/_apis/serviceendpoint/endpoints/serviceconnectionid"
get_params = {
"api-version": "7.1-preview.4"
}
get_headers = {
'Content-Type': 'application/json',
'Authorization': 'Basic Based64PAT',
}
get_response = requests.get(project_url, params=get_params, headers=get_headers)
secret_value = "test"
get_data_obj_mod = json.loads(get_response.content)
get_data_obj_mod["authorization"]["parameters"]["serviceprincipalkey"] = secret_value
ref_format = json.dumps(get_data_obj_mod, indent=2)
put_params = {
"api-version": "7.1-preview.4"
}
put_headers = {
'Content-Type': 'application/json',
'Authorization': 'Basic Based64PAT',
}
put_response = requests.put(project_url, params=put_params, headers=put_headers, data=ref_format)
print(put_response.text)
Update3:
def update_sc_key(secret_value, app_name, sc_dependence):
print(f"SC DEPENDENCE:",sc_dependence)
personal_access_token = get_pat()
bearer = get_bearer()
if sc_dependence == True:
pat_credentials = BasicAuthentication('', personal_access_token)
connection = Connection(base_url=organization_url, creds=pat_credentials)
service_connection_name = app_name[len('SP_') + app_name[len('SP_'):].find('_') - 4:]
print(service_connection_name)
service_endpoint_client = connection.clients.get_service_endpoint_client()
service_connections = service_endpoint_client.get_service_endpoints(project_name)
for service_connection in service_connections:
if service_connection.name == "RENEWAL_SC_TEST": # service_connection_name
print(f'SERVICE CONNECTION NAME:', service_connection.name)
service_connection_id = service_connection.id
print(f'SERVICE CONNECTION ID:', service_connection_id)
# Get Service Connection body
project_url = organization_url + project_name + "/_apis/serviceendpoint/endpoints"
try:
# GET RESPONSE
get_params = {
"endpointIds": service_connection_id,
"api-version": "7.1-preview.4"
}
get_headers={
"Authorization": "Bearer " + bearer
}
get_response = requests.get(project_url, params=get_params, headers=get_headers)
response = json.loads(get_response.text)['value']
get_response_tr = response[0]
if get_response.status_code == 200:
# PUT RESPONSE TO CHANGE PWD
url = "https://dev.azure.com/****/_apis/serviceendpoint/endpoints/" + service_connection_id
get_response_tr["authorization"]["parameters"]["serviceprincipalkey"] = secret_value
ref_format = json.dumps(get_response_tr, indent=2)
print(ref_format)
put_params = {
"api-version": "7.1-preview.4"
}
put_headers={
"Authorization": "Bearer " + bearer,
"Content-Type": "application/json"
}
put_response = requests.put(url, params=put_params, headers=put_headers, data=ref_format)
print(put_response.status_code)
print(put_response.text)
else:
print(f"Error: {get_response.status_code} - {get_response.text}")
except json.JSONDecodeError as e:
print("ERROR: expired bearer token")
else:
print(f'No Service Connection')