How to update an Azure Service Endpoint using python?

Question

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}

Answer 1

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