Created
November 2, 2020 16:21
-
-
Save FilBot3/ec425a5a6a249afb677e5a866e76aef1 to your computer and use it in GitHub Desktop.
A generic REST API Client for Azure DevOps.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- coding: utf-8 -*- | |
"""Azure DevOps REST API Client | |
A personally developed REST API Client for Azure DevOps. | |
.. _Azure DevOps REST API Documentation: | |
https://docs.microsoft.com/en-us/rest/api/azure/devops/ | |
""" | |
import base64 | |
import configparser | |
import json | |
import os | |
import pprint as pp | |
import sys | |
import requests | |
def get_api_token(config_file_location: str = os.environ.get('HOME') + '/.local/azdo_config.ini'): | |
"""Get the API Token of the user | |
The config file, $HOME/.local/azdo_config.ini looks like this: | |
[auth] | |
pat = yourpatgoeshere | |
Then this will read that file in. | |
Parameters | |
---------- | |
config_file_location : str, optional | |
The location with the Azure DevOps PAT in INI format. See ConfigParser | |
Returns | |
------- | |
str | |
The Azure DevOps PAT | |
""" | |
config = configparser.ConfigParser() | |
config.read(config_file_location) | |
return config['auth']['pat'] | |
# pylint: disable=line-too-long,too-many-arguments,too-many-locals,too-many-branches | |
def client(collection: str = None, base_url: str = "dev.azure.com", | |
project: str = None, team: str = None, area: str = None, | |
resource: str = None, item: str = None, | |
api_version: str = "5.1-preview", | |
api_method: str = "GET", | |
message_data: dict = None, | |
raw_url: str = None, | |
query_params: list = None) -> dict: | |
"""Performing requests to Azure DevOps | |
Parameters | |
---------- | |
collection : str | |
The Azure DevOps Collection or Top Account. | |
project : str, optional | |
The project in which to perform API calls. | |
team : str, optional | |
Some APIs are called against a Team, like work. | |
area : str | |
This is the "security namespace" or Azure DevOps component | |
to reach out to. Some Areas have resources to add on. | |
resource : str, optional | |
This is the resource within the area as listed in the docs. | |
item : str, optional | |
Some resources have an item listed with them. Sometimes that | |
item also has some other parts to it, reference the API. If so, simply | |
add those on to the end of the item like: item="170040/comments" | |
This then will be in-line with the API Resource Path. | |
api_version : str, optional | |
Individual components of the ADO REST API change and are given | |
different API Version numbers. It's quite annoying. | |
api_method : str, optional | |
The REST Method to perform. Should follow standard REST Methods and also | |
depends on what the Azure DevOps REST API calls for. | |
message_data : dict, optional | |
The Dict of values to send to Azure DevOps. This Dict will be turned into | |
JSON that will be part of the POST, PUT, PATCH message. | |
raw_url : str, optional | |
This is an overriding parameter that overrides the built resource_path | |
so that you can specify which URL's to use instead of letting the logic | |
here try to figure that out. | |
This will need to include the whole bits from /_apis/...?api-version=... | |
query_params : List<str>, optional | |
This allows the user to specify query parameters that normally come after | |
..?param=value¶m=value in a URL request. Then it will tack on the | |
api_version at the end of the string. If not, this is ignored and life | |
goes on. | |
Returns | |
------- | |
dict | |
Returns the JSON output from Azure DevOps as a Python Dictionary. | |
If an error occurs, that error is printed, and then returned due to how | |
the requests library works. | |
{ | |
'status_code': <requests.status_code:int>, | |
'data': <optional,requests.json(),dict> | |
} | |
.. _Azure DevOps REST API Documentation | |
https://docs.microsoft.com/en-us/rest/api/azure/devops/ | |
""" | |
username = "" | |
userpass = username + ":" + get_api_token() | |
b64_user_pass = base64.b64encode(userpass.encode()).decode() | |
organization_url = f"https://{base_url}" | |
if collection is not None: | |
organization_url += f"/{collection}" | |
if project is not None: | |
# Mostly because I don't know what to do instead of this currently. | |
organization_url += f"/{project}" | |
if team is not None: | |
organization_url += f"/{team}" | |
resource_path = f'/_apis' | |
if area is not None: | |
resource_path += f'/{area}' | |
if resource is not None: | |
resource_path += f'/{resource}' | |
if item is not None: | |
resource_path += f'/{item}' | |
if query_params is not None: | |
resource_path += '?' | |
for param in query_params: | |
resource_path += f'{param}&' | |
resource_path += f'api-version={api_version}' | |
else: | |
resource_path += f'?api-version={api_version}' | |
if raw_url is not None: | |
resource_path = None | |
resource_path = raw_url | |
headers = { | |
'Authorization': 'Basic %s' % b64_user_pass, | |
'Content-Type': 'application/json' | |
} | |
try: | |
if api_method == "GET": | |
ado_response = requests.get( | |
organization_url + resource_path, headers=headers) | |
elif api_method == "POST": | |
ado_response = requests.post( | |
organization_url + resource_path, headers=headers, data=json.dumps(message_data)) | |
elif api_method == "PUT": | |
ado_response = requests.put( | |
organization_url + resource_path, headers=headers, data=json.dumps(message_data)) | |
elif api_method == "PATCH": | |
ado_response = requests.patch( | |
organization_url + resource_path, headers=headers, data=json.dumps(message_data)) | |
elif api_method == "DELETE": | |
ado_response = requests.delete( | |
organization_url + resource_path, headers=headers, data=json.dumps(message_data)) | |
elif api_method == "HEAD": | |
ado_response = requests.head( | |
organization_url + resource_path, headers=headers) | |
else: | |
ado_response = requests.get( | |
organization_url + resource_path, headers=headers) | |
ado_response.raise_for_status() | |
client_response: dict = {} | |
try: | |
client_response['data'] = ado_response.json() | |
except ValueError: | |
print('No JSON returned.') | |
client_response = ado_response.json() | |
client_response.update({'headers': ado_response.headers}) | |
return client_response | |
except requests.exceptions.HTTPError as err: | |
# If an error occurs, drop out of the script. | |
sys.exit(err) |
you should rethink how many if statements you are using...
Try a switch statement next time
There is no direct switch/case statements in Python.
you should rethink how many if statements you are using...
How would you handle this then? This is the only way I knew how to do it at the time and would be open to suggestions.
dynamic dispatch is the keyword, look at how much copy and paste you have... should make it obvious that there must be a better way. btw if api_method == "GET": and the else block are the same
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Try a switch statement next time