debugpythongitMinor
Elegantly handle github API requests exceptions
Viewed 0 times
exceptionshandlegithubelegantlyrequestsapi
Problem
I have this code which creates a team using the github API and returns the team-id(success) or -1(failure).
My problem is with the error handling. When the request responses with an error status code I try to deal with a few different types of failure. It looks like a mess!
Is there any way to make it more readable/elegant?
I'd also like to know if I am catching all the reasonable exceptions that can be thrown by requests.
In case you were wondering why I don't just check
My problem is with the error handling. When the request responses with an error status code I try to deal with a few different types of failure. It looks like a mess!
Is there any way to make it more readable/elegant?
I'd also like to know if I am catching all the reasonable exceptions that can be thrown by requests.
def create_team(org,team_name):
'''
Try to create a team and return its id
if already exisits return id
'''
try:
data = {
"name": team_name,
"permission": "push"
}
headers = {'Authorization': 'token '+OAUTH_TOKEN}
response = requests.post(GH_API_URL+'orgs/'+org['login']+'/teams',
headers=headers,
data = json.dumps(data))
response.raise_for_status() #throw exception if request does not retun 2xx
#http status is 2xx, team must have been created
json_response = json.loads(response.text)
return json_response['id'] #get id of new team from response
except requests.exceptions.HTTPError as e:
if response.status_code == 422: #Unprocessable Entity
json_response = json.loads(response.text)
if json_response['errors'][0]['code'] == 'already_exists': #Unprocessable because already exists
print(' not created, team already exists!!!')
#get the id of existing team
team_id = get_teamID_from_name(org, team_name)
return team_id
else: #Unprocessable for some other reason
print(' HTTP error: '+str(e))
else: #other HTTP error != 422
print(' HTTP error: '+str(e))
except requests.exceptions.RequestException as e:
print('Connection error: '+str(e))
return -1In case you were wondering why I don't just check
Solution
One way to slightly speed-up is by reducing the overhead of the
You would have to start it with a context-manager in you outer function (where you loop over all of the team-names to create) and pass the session object to the
I also shuffled some code out of the
I also tried including a bare bones implementation for your outer function, but yours might be slightly different, of course.
I used
requests module. Currently your code has to establish a connection once for every team, re-transmitting the authentication API code and so on. This is a solved problem, you could use sessions.You would have to start it with a context-manager in you outer function (where you loop over all of the team-names to create) and pass the session object to the
create_team function.I also shuffled some code out of the
try block, you should have only the least possible amount in there, so you can be sure what exactly raised an error. I also inserted early returns, but there are other schools of thought which prefer a single exit for functions at the cost of having more code in the try block.I also tried including a bare bones implementation for your outer function, but yours might be slightly different, of course.
I used
str.format, where applicable instead of string addition.import requests
OAUTH_TOKEN = "XXX"
HEADERS = {'Authorization': 'token {}'.format(OAUTH_TOKEN)}
GH_API_URL = "http://github.com/XXX"
def create_teams(org, team_names):
with requests.Session() as session:
session.headers.update(HEADERS)
for team_name in team_names:
create_team(session, org['login'], team_name)
def create_team(session, org_name, team_name):
'''
Try to create a team and return its id.
If it already exists return id
'''
data = {
"name": team_name,
"permission": "push"
}
url = '{}orgs/{}/teams'.format(GH_API_URL, org_name)
try:
response = session.post(url, data=json.dumps(data))
response.raise_for_status() # throw exception if request does not return 2xx
except requests.exceptions.HTTPError as e:
if response.status_code == 422: # Unprocessable Entity
json_response = json.loads(response.text)
# Unprocessable because already exists
if json_response['errors'][0]['code'] == 'already_exists':
print('{} not created, team already exists!!!'.format(team_name))
# get the id of existing team
return get_teamID_from_name(session, org, team_name)
# Unprocessable for some other reason or other HTTP error != 422
print(' HTTP error: {}'.format(e))
return -1
except requests.exceptions.RequestException as e:
print('Connection error: {}'.format(e))
return -1
# http status is 2xx, team must have been created
json_response = json.loads(response.text)
return json_response['id'] # get id of new team from responseCode Snippets
import requests
OAUTH_TOKEN = "XXX"
HEADERS = {'Authorization': 'token {}'.format(OAUTH_TOKEN)}
GH_API_URL = "http://github.com/XXX"
def create_teams(org, team_names):
with requests.Session() as session:
session.headers.update(HEADERS)
for team_name in team_names:
create_team(session, org['login'], team_name)
def create_team(session, org_name, team_name):
'''
Try to create a team and return its id.
If it already exists return id
'''
data = {
"name": team_name,
"permission": "push"
}
url = '{}orgs/{}/teams'.format(GH_API_URL, org_name)
try:
response = session.post(url, data=json.dumps(data))
response.raise_for_status() # throw exception if request does not return 2xx
except requests.exceptions.HTTPError as e:
if response.status_code == 422: # Unprocessable Entity
json_response = json.loads(response.text)
# Unprocessable because already exists
if json_response['errors'][0]['code'] == 'already_exists':
print('{} not created, team already exists!!!'.format(team_name))
# get the id of existing team
return get_teamID_from_name(session, org, team_name)
# Unprocessable for some other reason or other HTTP error != 422
print(' HTTP error: {}'.format(e))
return -1
except requests.exceptions.RequestException as e:
print('Connection error: {}'.format(e))
return -1
# http status is 2xx, team must have been created
json_response = json.loads(response.text)
return json_response['id'] # get id of new team from responseContext
StackExchange Code Review Q#156525, answer score: 4
Revisions (0)
No revisions yet.