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

Recap

Last session we:

1. Created a ProvisionNewUser lambda in AWS
2. Created a requests layer and attached it to our lambda
3. 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:

1. Updating our ProvisionNewUser lambda to configure webhooks
2. Creating a lambda to process a webhook
3. 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!