I'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 are here. 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 called apis 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 json
from datetime import datetime
import requests
from django.core import signing
from environ import Env
from apis.models import XeroAccessToken
env = 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
loggingand so how do you know what happens to your app except for what you present in the client? \$\endgroup\$