DEV Community: bradenrichardson

Creating an Automated Personal Finances Dashboard with AWS (Part 6) - CDK

bradenrichardson — Sat, 08 Jan 2022 23:53:41 +0000

The final installment of this series details wrapping our AWS resources in some CDK (Cloud Development Kit) magic.

*Disclaimer: The scope of CDK does not include the Athena Workgroup or Athena DynamoDB connector - this will be addressed at a later point

I'll be honest, before this blog series I had only used Cloudformation with YAML files, and while I knew about CDK, I didn't know just how good it would be.
Developing in Cloudformation can be cumbersome, and it can impede the creative process that is so necessary when sandboxing, that's why I had it scheduled at the end: After all the fun stuff is done we go through and define our resources, making sure that it all deploys correctly.

During this segment I realised something: I should have been using CDK to develop from the start. It actually reduces development time and complexity by implementing best practice architecture by default, and filling in the gaps for you. Let's have a look at the differences between declarative Cloudformation and CDK.

CDK is object based, Cloudformation is declarative
Cloudformation needs 100% of the information, CDK will fill in the gaps (with best practice architecture)
Cloudformation requires a deployment mechanism built around it for automated workloads, CDK has it's own CLI and tooling

Process

To turn our deployment into a CDK workload I did the following:

Created a CDK project following this resource: https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html

Created an object per resource required
Assigned privileges to the resources
Deployed!

It was that easy, CDK handles unique bucket names, IAM roles and policies, a whole bunch of stuff that is normally tedious and boring to develop.

Here's the code block as it stands:

class CdkStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        dynamodb = _dynamo.Table(self, 'cdk_test',
            partition_key=_dynamo.Attribute(name='TransactionID', type=_dynamo.AttributeType.STRING) 
        )

        requests_layer = _lambda.LayerVersion(self, 'requests_layer',
            code=_lambda.Code.from_asset('layers\_requests'),
            compatible_runtimes=[_lambda.Runtime.PYTHON_3_9],
            compatible_architectures=[_lambda.Architecture.X86_64],
            description='A layer to send API requests'
        )

        process_webhook = _lambda.Function(self, 'process_webhook', 
            runtime=_lambda.Runtime.PYTHON_3_9, 
            handler='lambda_function.handler',
            code= _lambda.Code.from_asset('lambdas\process_webhook'),
            layers=[requests_layer]
            )

        provision_user = _lambda.Function(self, "provision_user",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler='lambda_function.handler',
            code= _lambda.Code.from_asset('lambdas\provision_user'),
            layers=[requests_layer]
        )

        dynamodb.grant_read_write_data(process_webhook)

        dynamodb.grant_read_write_data(provision_user)

        process_webhook_integration = HttpLambdaIntegration("Process Webhook Integration", process_webhook)

        api = aws_apigatewayv2.HttpApi(self, "HttpApi")

        api.add_routes(
            path="/webhook",
            methods=[aws_apigatewayv2.HttpMethod.POST],
            integration=process_webhook_integration
        )

Results

Here's a screenshot of the CDK stack deploying succesfully:

And here's the current state of what we've developed so far:

That brings a wrap to this blog series, I hope that you learnt a few things about AWS and how you can apply it the problems you encounter in your own life, I certainly learnt a lot along the way!

I'll be continuing development of this project in DIY Dashboard, an open source repo designed to help you quickly stand up serverless infrastructure to create your own automated dashboards. There are many planned features and a few refactors on the cards as well, I need to sort out that pesky provision_new_user lambda!

Creating an Automated Personal Finances Dashboard with AWS - Part 5 (Quicksight)

bradenrichardson — Tue, 04 Jan 2022 07:04:02 +0000

G'day everyone, in the last post DynamoDB was configured and was receiving realtime transaction history.

In this post we get down to the really good stuff, creating a personal finances dashboard!

Quick side note: The purpose of this guide is technical rather than financial, so the underlying financial data has been scrambled.

Prerequisite: Getting DynamoDB data into Quicksight

For this section I will defer to a fantastic article from another dev.to poster:
Using Athena data connectors to visualize DynamoDB data with AWS QuickSight

Quicksight Configuration

Alright so now that we've got our data from DynamoDB accessible by Quicksight, it's time to create a dataset.

Create a new data set from an Athena data source
Select the catalog that you created in the Athena guide
Select the table that we created in our DynamoDB guide
Hit visualize
Make sure to schedule a refresh of the data, this should align with your use case

So now we have a default analysis created from a dataset, the fun begins.

Review Use Cases

It's a good time to review our use cases for the dashboard, these will drive the visualisations that we create.

To deep dive historical data and identify correlations
To gain insight into current financial state
To forecast future financial state and identify trends

At a first glance I think the first two are achievable in a simplified state given our current technical capabilities, the forecasting aspect will need to be addressed in a future revision. Let's break down those first two with some ideas on how we will achieve them.

Use Case: Historical data

Create visualisations
- Date, value, description, category
- 12 month and 1 week view

Use Case: Current financial state

Create weekly report
- Week in review dashboard
Create alert
- Daily spending threshold

Create Visualisations

There is an endless amount of possibilities when it comes to visualising data, but only a few that are useful.

Pie chart
- Break down spending by category
Line chart
- Break down spending by month
Table
- List transactions, ordered by date
KPI
- Track income/expenditure in key areas
- Conditional formatting indicates 'Good', 'Bad' and 'Warning' levels of spending

12 Month Dashboard

In this image we can see a simple dashboard that I've created, I'll create another one for the 1 week view that we will use for the weekly report. Duplicating sheets and visualisations in Quicksight is quick and intuitive - you'll find yourself using it frequently as it also copies across the filters.

1 Week Dashboard

Not pictured - I included a daily KPI for spending that we will use for our spending alert.

Creating reports and alerts

Now that we've created a baseline 12 month and 1 week dashboard, we can publish the analysis. This allows us greater functionality in the alerting and reporting space.

Reports

To create a recurring report, navigate to the dashboard you published and select 'Share' > 'Email Report'

I've chosen to send a report weekly at 9am Monday morning

Formatted for desktop and selecting the weekly sheet

Select the email recipients

Alerts

To create an alert you need to have a KPI visualisation, the rest is simple.

I've created an alert that emails me when my spending goes over $70/day in this example

At this point you will have:
- Athena data source configured
- Quicksight configured
- Analysis and visualisations created
- Reporting and alerts configured

Awesome, the dashboard works - the next step is to wrap everything in CDK to make sure that we can continue developing the platform in an efficient way.

Creating An Automated Personal Finances Dashboard - Part 4 (DynamoDB)

bradenrichardson — Mon, 20 Dec 2021 07:57:38 +0000

I wrote some BAD code and redesigned on the fly...

In the last post we had written the code to process a webhook event and set up the API infrastructure to support that.
This part was supposed to cover writing to a CSV file in an S3 bucket - well that has definitely changed. I have redesigned to support DynamoDB and honestly, it was what I should have done from the start.

Here's what the new system looks like:

Why not CSV in S3?

It's very hard to append to CSV files in S3, the functionality with Boto3 just isn't there
The solution would have required more engineering and maintenance
Not as cool

Why DynamoDB?

Fully managed by AWS
Fantastic Boto3 support
Will scale with the system, CSV in S3 would have only been practical for so long

Setting up DynamoDB

This was my first time using DynamoDB and it was really easy to set up, let's go through it:

Create a table in DynamoDB - this is all very self explanatory, I'll be using the TransactionID as the partition key.

... That's it - almost seems too easy right? There's no tricks here, this is what a fully managed NoSQL database looks like. It's just very easy.

Writing to DynamoDB

This is where things get a little more interesting, although still quite simple. The objective is to modify both of our lambdas to write their respective payloads directly do DynamoDB.

Before we can attempt to write to DynamoDB, we need to make sure we have the correct permissions, now someone will probably say "this is your bad code right here" and they would be correct. Blanket "allow all" policies are bad, and I should feel bad. There is a security review staged for the end of v0.1 where I'll go through and ensure all policies are least privileged compliant - right now I'm purely focused on making things work.

"Effect": "Allow",
"Action": [
    "dynamodb:*"

Now that we have granted our lambdas supreme DynamoDB control we can write some code. The boto3 support for DynamoDB is great, it's really easy to use and super intuitive. One of my favourite things about DynamoDB is that you don't have to define a schema before hand, at the minimum you need a partition key, and then you can just add any other key value pair that you want. 'S' and 'N' denote string or number values.

This is the function that the process webhook lambda uses:

def write_to_dynamo(dictionary):
    dynamodb = boto3.client('dynamodb')
    dynamodb.put_item(TableName='quicksightTest', Item={'TransactionID':{'S': dictionary['ID']},'Category':{'S': dictionary['Category']}, 
    'ParentCategory' : {'S' : dictionary['ParentCategory']}, 'Value' : {'N' : dictionary['Value']}, 'Description' : {'S' : dictionary['Description']}, 
    'CreatedAt' : {'S' : dictionary['CreatedAt']}})

The Bad Code

Is code bad if it works? Probably yes.
Why was it bad? A lack of concurrency and an overall inefficient design.

I replicated the process webhook code for the provision new user lambda. Makes sense right? It does basically the same thing just on a larger scale

Here's the code block as it stands:

def write_to_dynamo(dictionary):
    dynamodb = boto3.client('dynamodb')
    a = 0
    for transaction in dictionary['id']:
        dynamodb.put_item(TableName='quicksightTest', Item={'TransactionID':{'S': dictionary['id'][a]},'Category':{'S': dictionary['category'][a]}, 
        'ParentCategory' : {'S' : dictionary['parentCategory'][a]}, 'Value' : {'N' : dictionary['value'][a]}, 'Description' : {'S' : dictionary['description'][a]}, 
        'CreatedAt' : {'S' : dictionary['createdAt'][a]}})
        a += 1


def lambda_handler(event, context):
    dictionary = create_Dictionary()
    write_to_dynamo(dictionary)

When I first tried running this, it timed out. Nothing to worry about, the lambda limit was only set to a minute, a large transaction history will take longer than that to download.

In the end I had to increase the lambda timeout to it's max of 15 minutes, and my lambda resource ran for ~13 minutes. This is clearly not ideal as my transaction history will only grow in size, meaning there will come a time when this code just does not work.

So what did I do?

Nothing.

If this was on the process webhook side I would have been forced to refactor and create a more efficient solution, but this function is only run once to provision a new user, it can afford to be inefficient for now.

Even though it's ugly, it worked, I now have 4,418 records in DynamoDB and more being added with every webhook event. Although it got me thinking about how I'm handling the compute, there are many ways I can improve the whole package. Stay tuned for those updates.

Today's Progress

Created a table in DynamoDB
Wrote a function that writes to DynamoDB
Imported my entire transaction history to DynamoDB!!!

Next up we will go through querying dynamoDB with Athena, and bringing our financial records into Quicksight - this is the fun stuff!

Creating An Automated Personal Finances Dashboard With AWS - Part 3 (Webhook Lambda)

bradenrichardson — Wed, 15 Dec 2021 13:08:12 +0000

Recap

Last session we:

Created a ProvisionNewUser lambda in AWS
Created a requests layer and attached it to our lambda
Wrote some Python code that sends a GET request to Up's API endpoint to download our historical data

Now that we can successfully pull our historical data, we need to look to our realtime data. To do this we need to leverage Up's webhooks, in this scenario, Up's webhook sends a POST request to an API endpoint whenever there is a new transaction on the bank account.

Do the thing

First up we need to create a new lambda that will handle our webhooks.

Let's just put some dummy code in here that prints out the event, this will be helpful later when our lambda is being triggered and we want to see the structure of the request.

def lambda_handler(event, context):
print(event)

Create an API gateway

We need to specify an integration for our API to communicate with, API gateway has native support for lambda so this whole process is really easy

Specify a gateway route

Our route needs to correspond to it's functionality, we are expecting our API gateway to receive a POST request. The resource path is appended to our endpoint url.

Create a stage
Default values for this one.

Now things start to move around a bit, we need to register a webhook with Up's API - this sounds like a job for our ProvisionNewUser lambda:

def create_webhook(invoke_url):
    api_url = api_url_base + 'webhooks' 
    data_object = {"data": {"attributes": {"url" : invoke_url}}}
    response = requests.post(api_url, headers=headers, json=data_object)
    print(response.text)

Right now we don't have any way of actually telling the ProvisionNewUser lambda when to register a webhook and when to download a historical data report - this will all change, to start with we will manually call the create_webhook() function from within the lambda.

Our webhook has been configured, let's test it out by transferring some money within the app. This should send our API gateway a post event, triggering our ProcessWebhook lambda.
Here's the result:

This is where our print statement comes in handy, the original Up webhook payload will be wrapped with the API gateway meta data, now we can build a statement to access the data that we need.

The data we get from this payload isn't actually the information we want, it's just a transaction ID, we need to go back to Up's API and request the rest of the details using that transaction ID.

def retrieve_transaction(transaction_id):
    api_url = api_url_base + 'transactions/' + transaction_id 
    response = requests.get(api_url, headers=headers)
    data = response.json()
    dictionary = {
        'ID' : transaction_id,
        'Description' : data.get('data').get('attributes').get('description'),
        'Value' : data.get('data').get('attributes').get('amount').get('value')[1:],
        'Created At' : data.get('data').get('attributes').get('createdAt')
    }
    if data.get('data').get('attributes').get('amount').get('value') < 0:
        pass
    if data.get('data').get('relationships').get('category').get('data'):
        dictionary['Category'] = data.get('data').get('relationships').get('category').get('data').get('id')
    else:
        dictionary['Category'] = 'Uncategorized'
    if data.get('data').get('relationships').get('parentCategory').get('data'):
        dictionary['Parent Category'] = data.get('data').get('relationships').get('parentCategory').get('data').get('id')
    else:
        dictionary['Parent Category'] = 'Uncategorized'
    return dictionary

We're requesting data again - this requires the requests layer that we previously built, but this time applied to the ProcessWebhook lambda.

Currently this function doesn't actually do anything, we need to insert the following into the lambda handler:

transaction_id = event.get('body').get('data').get('relationships').get('transaction').get('data').get('id')
transaction = retrieve_transaction(transaction_id)

Rather than always triggering the webhook, we have captured the JSON payload once before so we can just reuse that. Create a test event in lambda and paste in the previous payload you received, this will make development much easier!

Test the lambda with the test event

Finally we're ready to see how our lambda would react to a webhook

Great stuff! It recognised that I transferred $7.88 to savings, this was the original webhook payload that came through earlier. Now whenever a webhook payload hits our API, our lambda will automatically request the contents of that transaction ID and return it as a dictionary.

Summary

In this post we covered:

Updating our ProvisionNewUser lambda to configure webhooks
Creating a lambda to process a webhook
Creating API infrastructure to trigger a lambda

Next Steps

Up next we cover writing to a csv file in an S3 bucket, not long now until we can visualize our data!

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

bradenrichardson — Tue, 14 Dec 2021 12:12:32 +0000

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!
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?

There are a couple of ways to solve this problem, my favourite one is to create a lambda layer, as these are then reusable across your AWS account with other functions.

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.

Awesome - It's hitting the Up API endpoint and processing tokens!

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!

Creating An Automated Personal Finances Dashboard With AWS - Part 1 (Vision)

bradenrichardson — Tue, 14 Dec 2021 11:09:34 +0000

I like keeping track of things.

In fact, I like keeping track of things so much that I joined a bank that gave me in depth analytics and insights into my spendings and savings (UP Bank).

Up is a pretty new bank here in Australia and they've managed to create a really cool banking platform with a nifty app. Perhaps my biggest gripe though is that there is no desktop application, and their insights and analytics aren't very customizable and there aren't any out of the box 3rd party integrations.

I had caught the UP bug and I wanted even deeper insight into my finances, I wanted to be able to piece together correlations and look at long term trends.

And then UP released their API, an answer to my prayer. The seed was planted and it was time to get to work.

Within days I had built a python script that downloaded my entire transaction history to a csv file, I would then import that CSV into Quicksight manually. This was amazing, the rush was unreal, but after I had to repeatedly perform the manual steps over and over I knew this wouldn't work long term, it was crying out to be automated.

Why do I want an automated financial dashboard?

To gain insight into current financial state
To deep dive historical data and identify correlations
To forecast future financial state and identify trends

How am I going to do this?

Good question, right now I actually don't know what the end product will look like.

Instead I've chosen to identify an MVP (minimum viable product) and to try to replicate my original workflow as close as possible, but substituting manual tasks for automated ones. This way I get the most value in the quickest way. After I've delivered v0.1 I can then look at the next jump that will provide the most value from the least amount of effort.

Let's expand on a few core decisions that have been made:

AWS - My cloud provider of choice
Lambda - Severless makes the most sense when lifting and shifting python scripts
S3 - This also most closely represents the current workflow, perfect for an MVP
Cloudformation - Baking everything into a cloudformation script will allow us to easily grow the solution without having to worry about cleaning up leftover or misconfigured resources in AWS

Next steps

Stay tuned for the next part which will cover creating and configuring the Provision User lambda function!