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

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.

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

Create an API gateway

Specify a gateway route

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!