6
\$\begingroup\$

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

\$\endgroup\$
2
  • \$\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 logging and so how do you know what happens to your app except for what you present in the client? \$\endgroup\$ Sep 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
    Sep 7, 2020 at 10:39

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.