Creating An Automated Personal Finances Dashboard With AWS - Part 2 (Request Lambda)

G'day folks, last time we spoke we covered the following:

1. Pain points of my manual solution
2. Use cases of an automated solution
3. Basic MVP design

In this post we're going to tackle the first obstacle, the Provision New User lambda.

Now these two lambdas could be packaged as a single lambda, with separate functions for each use case, however for the sake of simplicity we are going to split these out, and in the future they will probably end up together in some fashion.

Lambda Overview:

1. Provision New User
   • Sends a GET request
   • Manually triggered
   • Normally only used once or when data needs to be completely refreshed
2. Process Webhook
   • Receives a POST request
   • Triggered every time a transaction is registered through a webhook (we'll get to that later)
   • This is the core of the automated dashboard and will provide real time data

Plan Of Action

1. Create a lambda in AWS
2. Write some python code that sends a GET request to Up's API endpoint and stores the transaction history in a dictionary

Sounds simple enough right?

Let's crack on with it.

Creating a lambda function in AWS

1. Navigate to the AWS console and create a new Lambda function, we're choosing Python 3.9 as our runtime and leaving everything else default. Note the default execution role, this will come up later when we look at deploying our solution with Cloudformation!
2. Write some code to hit the UP Api endpoint
   • This code filters through the response and grabs all of the relevant info, putting it all into a dictionary of arrays

import json
import os
import requests

api_token = os.getenv('api_token')
api_url_base = 'https://api.up.com.au/api/v1/'
headers = {'Authorization': 'Bearer {}'.format(api_token)}

def create_list(api_url):
    response = requests.get(api_url, headers=headers)
    if response.status_code == 200:
        data = []
        data.append(response.json().get('data'))
        if response.json().get('links').get('next'):
            token = response.json().get('links').get('next')
            while token:
                response = requests.get(token, headers=headers)
                data.append(response.json().get('data'))
                token = response.json().get('links').get('next')
                if token:
                    print("Processing token: {}".format(token))
                else:
                    print("Finished processing tokens")
        return data
    else:
        print(response.status_code)

def create_csvDictionary():
    api_url = api_url_base + 'transactions'
    data = create_list(api_url)
    csvDictionary = {'id' : [], 'description' : [], 'value' : [], 'category' : [], 'parentCategory' : [], 'createdAt' : []}

    for array in data:
        for transaction in array:
            if 'Transfer' in transaction.get('attributes').get('description'):
                continue
            if 'transfer' in transaction.get('attributes').get('description'):
                continue
            if 'Cover' in transaction.get('attributes').get('description'):
                continue
            if 'Round Up' in transaction.get('attributes').get('description'):
                continue
            if float(transaction.get('attributes').get('amount').get('value')) > 0:
                continue
            else:
                csvDictionary['id'].append(transaction.get('id'))
                csvDictionary['description'].append(transaction.get('attributes').get('description'))
                csvDictionary['value'].append(transaction.get('attributes').get('amount').get('value')[1:])
                if transaction.get('relationships').get('category').get('data'):
                    csvDictionary['category'].append(transaction.get('relationships').get('category').get('data').get('id'))
                else:
                    csvDictionary['category'].append('Uncategorized')
                if transaction.get('relationships').get('parentCategory').get('data'):
                    csvDictionary['parentCategory'].append(transaction.get('relationships').get('parentCategory').get('data').get('id'))
                else:
                    csvDictionary['parentCategory'].append('Uncategorized')
                csvDictionary['createdAt'].append(transaction.get('attributes').get('createdAt'))

    print(csvDictionary)
    return csvDictionary


def lambda_handler(event, context):
    create_csvDictionary()

Our first roadblock! Our lambda doesn't have the requests library, how are we going to sort this one out?

Creating a lambda layer

1. Install the package locally (specify --no-user tag if on Windows)

pip3 install requests --target .\requests --no-user

1. Zip the package
2. Upload to the Lambda Layer console, ensuring your runtime is correct

1. Add the layer to your lambda function

1. Okay so now our lambda function has access to the request library, we're going to try this block of code out and see how it goes.

Not Awesome - The lambda initially timed out because there is so much data there (2 years worth of transactions)

In Summary

1. We built a Lambda function that sends a GET request to an API endpoint and receives a response back
2. We created a Requests lambda layer to enable the GET request and attached it to our lambda
3. The Lambda function takes A LONG time to pull a historical dataset... let's revisit this in the future but for now - it works.

Next Time

Next post will be covering the Webhook (event based) lambda function and the corresponding API gateway and endpoint that will be required. Can't wait!