\$\begingroup\$\$\endgroup\$
2I'm not actually that new to writing Python. But I've no formal training and have only learnt to build applications to solve problems for my job out of necessity. Starting to bring my existing skills up to a professional standard is a long overdue project. I'd like to humbly post some snippets here for criticism to guide me in filling the gaping holes in my education.
This is also my first post, so I hope I'm not one of the 10-20 people a day who haven't read the rules correctly.
Context
The below is a working part of a Django project which will automate some workflows for me. The class in this snippet is used to interact with Xero, an accounting platform I use for my finances, via their API.
Docs for the Xero API arehere. The interaction with Xero is working, so my main concern is with the quality and layout of my work. However my error-handling feels shoddy so any pointers there would be amazing too.
I have a Django app calledapis in the project which includes Python files for interaction with specific external APIs. The classes within are instantiated from elsewhere in the project to complete various different workflows involving those external platforms. There are also very basic models in the app for storing access tokens between uses.
xero.py
import jsonfrom datetime import datetimeimport requestsfrom django.core import signingfrom environ import Envfrom apis.models import XeroAccessTokenenv = Env()class XeroApi: def __init__(self): pass # auth functions def get_auth_url(self): scope = 'offline_access ' \ 'openid ' \ 'accounting.transactions ' \ 'accounting.contacts ' \ 'accounting.settings' return ('https://login.xero.com/identity/connect/authorize' '?response_type=code' '&client_id={}' '&redirect_uri={}' '&scope={}' '&state={}').format( env('XERO__CLIENT_ID'), env('XERO__REDIRECT_URI'), scope, env('XERO__STATE') ) def request_token(self, auth_code): auth_credentials = base64.urlsafe_b64encode( bytes( '{}:{}'.format( env('XERO__CLIENT_ID'), env('XERO__CLIENT_SECRET') ), 'utf-8' ) ).decode('utf-8') headers = { 'Authorization': 'Basic {}'.format(auth_credentials) } payload = { 'grant_type': 'authorization_code', 'code': auth_code, 'redirect_uri': env('XERO__REDIRECT_URI') } return requests.post( 'https://identity.xero.com/connect/token', headers=headers, data=payload ) def refresh_token(self, access_credentials): auth_credentials = base64.urlsafe_b64encode( bytes( '{}:{}'.format( env('XERO__CLIENT_ID'), env('XERO__CLIENT_SECRET') ), 'utf-8' ) ).decode('utf-8') headers = { 'Authorization': 'Basic {}'.format(auth_credentials), 'Content-Type': 'application/x-www-form-urlencoded' } payload = { 'grant_type': 'refresh_token', 'refresh_token': access_credentials['refresh_token'] } return requests.post( 'https://identity.xero.com/connect/token', headers=headers, data=payload ) def get_tenant_id(self, access_token): url = 'https://api.xero.com/connections' headers = { 'Authorization': 'Bearer {}'.format(access_token), 'Content-Type': 'application.json' } response = requests.get(url, headers=headers).json() return response[0]['tenantId'] def retrieve_stored_token(self): access_token_objects = XeroAccessToken.objects.all() if len(access_token_objects): access_credentials = json.loads( signing.loads( access_token_objects[0].signed_access_credentials ) ) tenant_id = signing.loads(access_token_objects[0].signed_tenant_id) return access_credentials, tenant_id else: return False, False def update_access_credentials(self): credentials, tenant_id = self.retrieve_stored_token() if not credentials: return {'error': 'Access credentials not found.'} now = datetime.now() expires_at = datetime.fromtimestamp(credentials['expires_at']) if expires_at < now: credentials = self.refresh_token(credentials).json() credentials['expires_at'] = datetime.timestamp( now ) + credentials['expires_in'] - 30 credentials_object = XeroAccessToken.objects.all()[0] credentials_object.signed_access_credentials = signing.dumps( json.dumps(credentials) ) credentials_object.save() return { 'Authorization': 'Bearer {}'.format(credentials['access_token']), 'Xero-tenant-id': tenant_id, 'Accept': 'application/json' } # operational functions def get(self, endpoint, params={}): headers = self.update_access_credentials() if 'error' in headers: return {'failure': True, 'error': headers['error']} page = 1 more = True content = [] while more: url = 'https://api.xero.com/api.xro/2.0/{}?page={}'.format( endpoint, page ) response = self._clean_get( requests.get(url, headers=headers, params=params) ) if 'failure' in response: return response content += response[endpoint] page += 1 if len(response[endpoint]) < 100: more = False return {'failure': False, 'content': content} def post(self, endpoint, payload): headers = self.update_access_credentials() if 'error' in headers: return {'failure': True, 'error': headers['error']} url = 'https://api.xero.com/api.xro/2.0/{}'.format(endpoint) response = self._clean_post( requests.post(url, headers=headers, data=json.dumps(payload)) ) if 'failure' in response: return response return {'failure': False, 'content': response} def email_document(self, endpoint, id): headers = self.update_access_credentials() if 'error' in headers: return {'failure': True, 'error': headers['error']} url = 'https://api.xero.com/api.xro/2.0/{}/{}/Email'.format( endpoint, id ) response = self._clean_post(requests.post(url, headers=headers)) if 'failure' in response: return response return {'failure': False, 'content': response} # utils def _clean_get(self, response): if response.status_code > 299: return {'failure': True, 'status_code': response.status_code, 'error': response.reason} return response.json() def _clean_post(self, response): try: response = response.json() if 'ErrorNumber' in response: message = response['Message'] if 'Elements' in response: for element in response['Elements']: if 'ValidationErrors' in element: for error in element['ValidationErrors']: message += ' | {}'.format(error['Message']) return {'failure': True, 'status_code': response['ErrorNumber'], 'error': message} return response except ValueError: if response.status_code > 299: return {'failure': True, 'status_code': response.status_code, 'error': response.reason} return response
- \$\begingroup\$Welcome to CR! I don't really have the time for an answer right now but the first thing that came to mind are the missing error exceptions. What happens if a request throws an exception? Wouldn't that make your whole app crash? Also, there's no
loggingand so how do you know what happens to your app except for what you present in the client?\$\endgroup\$Grajdeanu Alex– Grajdeanu Alex2020-09-04 17:27:31 +00:00CommentedSep 4, 2020 at 17:27 - \$\begingroup\$Thanks @GrajdeanuAlex.! I have been trying to handle unexpected exceptions (asking for forgiveness) as part of the workflow in which this class is used, while this class focusses on expected issues returned from the external server (asking for permission). Appreciate the point about logging. The same applies as above with regard to the workflow using the class, but I think I could do with some more development in that area. I will do some learning today.\$\endgroup\$Simon– Simon2020-09-07 10:39:17 +00:00CommentedSep 7, 2020 at 10:39
You mustlog in to answer this question.
Explore related questions
See similar questions with these tags.