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

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

Pain points of my manual solution

Use cases of an automated solution

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:

Provision New User

Sends a GET request
Manually triggered
Normally only used once or when data needs to be completely refreshed

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

Create a lambda in AWS

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

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! Create Lambda Function

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

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

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

Zip the package

Upload to the Lambda Layer console, ensuring your runtime is correct

Add the layer to your lambda function

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

We built a Lambda function that sends a GET request to an API endpoint and receives a response back

We created a Requests lambda layer to enable the GET request and attached it to our lambda

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!