Forem: Taylor Facen

Finta's Automated Financial Synchronization powered by Plaid, Stripe and Inngest

Taylor Facen — Wed, 24 Apr 2024 05:36:00 +0000

Hi 👋🏽, I'm Taylor, founder and solo-builder of Finta. Finta helps users sync their financial data to their favorite apps, so that they can manage their finances how they want, where they want. The platform integrates with data providers like Plaid and Stripe to sync data to destination integrations like Notion, Airtable, Coda, and Google Sheets. These syncs can be triggered by webhooks when new data is available or by manual triggers initiated by users.

Inngest is the orchestration tool used to power this entire workflow. Essentially, Finta leverages Inngest to streamline and automate the backend processes, ensuring data synchronization is both accurate and efficient.

Problem Overview

There are two main problems that Inngest solves:

DRY Functions

Throughout the codebase, there are shared functions between multiple types of syncs. For example, if Plaid sends a “New Transactions” webhook, we need to use some sort of syncAccounts function before triggering a syncTransactions function. Also, if a user triggers a manual refresh, we'll need to use the same syncAccounts and syncTransactions functions. Inngest functions should be portable so that each sync trigger can call any mix of the appropriate sync functions in the correct order.

Function Orchestration

Certain sync steps need to happen in a specific order and need to resolve before the main syncing function can continue. For example, before syncing a user's holdings to their destination, we need to first sync all of the available securities data to their destination. Also, it's possible for a destination to have multiple running syncs. Imagine that a bank refreshes their transactions data at the same time that a user initiates a manual refresh. Even though syncs can happen concurrently, destinations should only have one sync function actively running at a time to avoid racing issues and duplicated data.

Key Inngest Features That We Use

There are two key Inngest features used for all syncing workflows at Finta:

`concurrency` with a key

To make sure a destination has at most one actively running sync function, each sync function has the following concurrency property:

const syncHoldingsFunction = inngest.createFunction({
    id: 'sync-holdings',
    name: 'Sync Holdings',
    concurrency: { limit: 1, key: 'event.data.destinationId' },
})

Because destinationId is a parameter for all sync functions, this guarantees that each function only runs once for a given destination. Also, when a sync completes, the next sync in line for that destination automatically starts.

`invoke`

This feature is the star of the show. Previously, we used waitForEvents to run functions synchronously.

const { destinationId, syncId } = event.data;

const transactions = await step.run(
  { id: 'fetch-transaction' }, 
  async () => ...
)

const syncTransactionsPromise = step.waitForEvent('app/destination.sync.task.finished', {
  timeout: '10m',
  if: `async.data.destinationId == '${destinationId}' && async.data.syncId == '${syncId}' && async.data.action == 'Sync' && async.data.table == 'Transactions'}`
})

await step.sendEvent({
name: 'app/destination.sync.transactions',
data: {
  destinationId,
  transactions,
  syncId
}
});

const syncTransactionsResponse = await syncTransactionsPromise;

However, thanks to invoke most of this logic can be replaced by a simple step.invoke() call. More about this will be explained further down below.

Implementation Feature 1: Composability (DRY Programming)

Before introducing Inngest, a single API route handled the entire end-to-end syncing flow in Finta. This would often fail for various reasons. As you can imagine, if a person has 10 bank connections each with over 1,000 transactions, it might take longer than the timeout limit to sync all of the data. Additionally, each banking integration and each destination integration has rate limits that need to be handled. Lastly, many sync triggers share the same steps. Thus, at one point in time, Finta's codebase looked like the following:

if (['stripe_balance_update', 'plaid_holdings_update', 'plaid_transactions_update', 'plaid_investments_update'].includes(syncTrigger) ) {
    await syncAccounts(...)
};

if (['stripe_transactions_update', 'plaid_transactions_update'].includes(syncTrigger) ) {
    await syncTransactions(...)
};

if (['plaid_holdings_update', 'plaid_investments_update'].includes(syncTrigger) ) {
    await syncSecurities(...)
};

if (['plaid_holdings_update'].includes(syncTrigger) ) {
    await syncHoldings(...)
};

if (['plaid_investments_update'].includes(syncTrigger) ) {
    await syncInvestments(...)
};
...
This was fine in the early days. However, now with over 20+ sync triggers, this would be hard to maintain.

Now, there are distinct sync events and functions for each data type. Each event requires a destinationId , a syncId , and the data in the event payload.

{
  name: 'app/destination.sync.accounts',
  data: {
    destinationId: string;
    syncId: string;
    accounts: Account[]
  }
}
Each event corresponds to a sync function:

const table = 'Accounts';

const syncAccounts = inngest.createFunction(
  {
    ...
  },
  async ({ event, step, utils }) => { // Utils comes from middleware
    const { destinationId, accounts } = event.data;
    const destination = await utils.fetchDestination(
      { id: 'fetch-destination' }, 
      destinationId
    );

    const { records, destinationError } = await utils.loadRecords(
      { id: 'load-account-records' }, {
        table,
        destinationId,
        destination
        }
    );

    if { destinationError } {
      return { hasError: true }
    }

    const currentAccounts = records.map(record => record.properties.id);
    const newAccounts = accounts.filter(account => !currentAccounts.includes(account.id));
    const updatedAccounts = accounts.filter(account => currentAccounts.includes(account.id));

    const createResults = await step.run(
      { id: 'sync-new-accounts-to-destination' }, 
      async () => {
        ...
      }
    );

    const updateResults = await step.run(
      { id: 'sync-updated-accounts-to-destination' },
      async () => {
        ...
      }
    );

    await utils.logResultsToDB({ id: 'log-results-to-db' }, { createResults, updateResults });

    return { hasError: false }
  }
)

Let's say that Finta decides to use a new data provider and needs to set up a new sync trigger event (for example, new_provider/accounts.updated ) to handle account updates. The corresponding Inngest function doesn't have to have any logic around syncing accounts. Instead, it can focus on fetching and formatting data from the new data provider and invoke the syncAccounts function as needed. The best part about all of this is that if any logic around how accounts sync to destinations needs to be changed, this logic only needs to be updated in one place.

Implementation Feature 2: Function Orchestration

There are multiple events that can trigger a sync. Each trigger event has its own Inngest function. Here is where the relevant data is gathered and formatted. Then the appropriate sync functions are called in the necessary order.

Event: plaid/transactions.default_update

Function:

import { syncAccounts, syncTransactions } from './functions/...';

const plaidDefaultUpdate = inngest.createFunction(
  {
    ...
  },
  { event: 'plaid/transactions.default_update' },
  async ({ event, step, utils }) => {
    const connectedDestinationIds = await utils.getConnectedDestinationIds(
      { id: 'get-connected-destination-ids' }, 
      data: event.data
    )

    const sync = await utils.startSync({ id: 'start-sync' }, ...);
    const accountsData = await step.run({ id: 'fetch-accounts-from-plaid' }, ...)

    const syncAccountsResponse = await Promise.all(connectedDestinationIds.map(destinationId => 
      step.invoke({ id: `sync-accounts-to-${destinationId}` }, {
        function: syncAccounts,
        data: {
          destinationid,
          syncId: sync.id,
          accounts: accountsData
        }
      })
    )).then(responses => {
      const hasError = Boolean(responses.find(response => response.hasError));

      return { responses, hasError }
    })

    if ( syncAccountsResponse.hasError ) {
      // Send notification
      // Update logs
      // etc.
      return;
    }

    const transactionsData = await step.run({ id: 'fetch-transactions-from-plaid' }, ...);
    const syncTransactionsResponse = ...
})

Because invoke resolves in a promise containing the results from the function, multiple syncs to different destinations can be triggered at once. Once all of them are done, the parent trigger function can handle the results and proceed accordingly.

Takeaways for you to use in your Inngest Project

Having used Inngest for over a year now, here are a few things that I've learned when it comes to building scalable and reliable architecture:

If you notice that many of your functions share the same logic, split that logic into a separate function and use invoke to trigger where needed
Although invoke solves most problems, there is still a need to use waitForEvent : if you are triggering a single function and waiting for that function to finish before proceeding, use invoke. If the event is triggered in a completely separate workflow and doesn't always happen at that exact moment, use waitForevent
With invoke and middleware functions, you can build complex workflows that follow DRY programming.

If you're building something similar or if you just want to chat about all things Inngest, feel free to reach out on LinkedIn or X.

I Built My Own Verzuz App

Taylor Facen — Fri, 01 May 2020 06:58:16 +0000

What I built

While social distancing, many artists and producers have been battling it out on Instagram to see who has the best music discography. Although Instagram is an awesome platform, it wasn't necessarily built for this kind of use case. Viewers can't vote on who they think won the round, no one can keep track of the score, and you can only see verified accounts in the comments section. This is why I build the Verzuz App. With this app, anyone can host their own battles and interact with all audience members in real time.

Category Submission: Exciting X-Factors

Demo Link

You can visit the live app here

Link to Code

Here's the link to the Github repo

How I built it (what's the stack? did I run into issues or discover something new along the way?)

At its core, the Verzuz app is a React App using Node Express and MongoDB. Additionally, here are some of the integrations within the app

Twilio Sendgrid - Sends an invite email to the players once the battle is created
Twilio Verify - Used to verify phone numbers to reduce spam
Pusher - Used for real time comments and events
Agora - Used for video broadcasting

The APIs themselves were fairly straightforward to implement. Although, I did struggle a bit on implementing the real-time comments and events in the right order. I learned that if you fetch data from the database and then subscribe to a websocket, then you could potentially miss some data. Also, I'm more of a backend person. So learning React and implementing design was a but of a struggle for me. The app currently isn't mobile responsive, but in the future that will change!

Additional Resources/Info

You can read more about the Verzuz battles on Timbaland or [Swizz Beats]{https://www.instagram.com/therealswizzz/?hl=en} Instagram accounts.

Also, battles are currently limited to 10 viewers as costs can rise pretty high for all of the services.

How I Used Twilio Autopilot to Build an Interview Coach

Taylor Facen — Sat, 22 Feb 2020 18:25:44 +0000

Link to the project

Link to the repo

This past October, I participated in the Women Make Just F*ing ship it challenge. Because the goal was to have a fully-made project by the end of the month, I figured this would be the perfect push to go through the entire ideation, development, and deployment process (I'm sure we can all relate to having a graveyard of Github repos for unfinished projects).

It was application season, so I decided to create a bot that can help users prep for interviews. For this project, I used a flask server via Python, PostgreSQL, and Twilio. I'm a huge fan of how easy it is to create logic-based conversations using Twilio’s Autopilot and thought this would be the perfect project to use it for. I deployed the app on Heroku (my favorite service for quickly launching an app).

Development

First I set up the server and connected it to the database. The two main models used are for the questions and categories.

Github commit

__init__.py

from flask import Flask
from flask_admin import Admin

from config import DATABASE_URL
from .db import db

def create_app():
    app = Flask(__name__)
    app.config['FLASK_ADMIN_SWATCH'] = 'paper'
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['SQLALCHEMY_DATABASE_URI'] = DATABASE_URL

    admin = Admin(name = "Interview Coach", template_mode = 'bootstrap3')

    admin.init_app(app)
    db.init_app(app)

    with app.app_context():
        from .models import Category, Question
        db.create_all()

    return app

models.py

from uuid import uuid4

from .db import db

class Base(db.Model):
    __abstract__ = True

    id = db.Column(db.String, primary_key = True, default = lambda: uuid4().hex)
    created_on = db.Column(db.DateTime, default=db.func.now())
    updated_on = db.Column(db.DateTime, onupdate=db.func.now())

    def to_dict(self):
        """Converts object to dictionary"""
        return {k: v for k, v in self.__dict__.items() if k[0] != '_'}


class Category(Base):
    """Category model for the different types of interview questions"""
    __tablename__ = 'categories'

    category = db.Column(db.String, unique = True)
    description = db.Column(db.Text)

class Question(Base):
    """Question model for storing information about interview questions"""
    __tablename__ = 'questions'

    question = db.Column(db.String)
    category_id = db.Column(db.String, db.ForeignKey('categories.id'))

The next step was to build in the conversation logic.

User initiates the conversation by texting something like “Practice” or “I want to practice”
The bot walks the user through selecting an interview question category (e.g. personal, behavioral, technical) and the amount of questions
The bot calls the user to start the interview. After asking randomly selected questions with the specified category, the bot stores the user’s responses for further analysis
When the conversation is done, the bot hangs up and sends a text message with feedback on talking speed, response length, and the amount of filler words. Then the user receives a picture of a word cloud of the most used words.

Lastly, I added in a password protected admin panel to CRUD data for the application.

Model views for categories and questions

And voila, now we have an Interview Coach bot!

One of the hardest parts about building this application was transferring the state of the conversation between the sms conversation, phone interview, and sms feedback. I didn’t want to store these details in the database, because this data seemed temporary (after the conversation is over, the data wouldn’t be needed in the future). To solve this, I used Twilio Sync as a datastore to hold information like the the category and number of questions. I used the user’s phone number as the distinct key and set the sync document to auto delete after an hour.

Lessons Learned

The biggest lesson was in learning how to be okay with releasing a minimum viable product (MVP). I originally had a laundry list of features that I wanted to incorporate, and overall, I felt extremely confident in my ability to code up each and every one of them. However, the goal was to get this project done in a month, and I didn't want my want for perfection to hurt my need for completion. So, I had to be honest with myself and pick the few features that were 100% required for me to call this project a success. It was hard but definitely worth it in the end.

Secondly, this was the first time I've completed the full project lifecycle on a personal project. I came up with an idea, mapped out the features, developed and tested the application, got user feedback, created a website, and launched a fully finished product. I have a new appreciation for how applications are built, and I feel like I have a better understanding of how each phase impacts each other.

What's Next?

Next up, I hope to integrate SendGrid to send the user the list of questions asked and their responses via email. I also want to add an analytics tool like Segment or Mixpanel to track activity to see the types and quantities of questions asked and how often a person uses the app. Stay tuned for updates, and in the meantime, you can check out the bot here.

If you don't already have one, sign up for a Twilio account. Using the referral link provided will give you $10 credit when you set up your account.

How to Safely Add Environment Variables to your Python App

Taylor Facen — Thu, 12 Dec 2019 03:53:07 +0000

Picture this. You have some awesome code that you want to share with the world. However, you don't want to or can’t share keys, credentials, and other sensitive information. Sure we can add these items to our bash_profile, but what if we accidentally use a set of keys for the wrong project? This is where the dotenv family saves the day. The dotenv module loads credentials from a .env file, allowing you to customize the settings for each project individually. For this sample console project, we’ll be using python-dotenv. There are also versions for Node and Ruby.

What we’re Building

This commit has the starter code to create an app that logs the weather to the console. The code is 99% done, and the only thing that’s left is to add the necessary API key to get it running. We’ll be using Dark Sky to retrieve weather data. So grab a key and head on over to your favorite text editor.

Let’s Code!

As you can see in the starter files, the api key is listed in plain site.

import requests

DARK_SKY_API_KEY = "my-super-secret-api-key"

def get_weather(latitude, longitude):
    url = "https://api.darksky.net/forecast/{}/{},{}".format(DARK_SKY_API_KEY, latitude, longitude)

    response = requests.get(url)
    data = response.json()

    current_weather = data['currently']
    summary = current_weather['summary'].lower()
    temperature = current_weather['temperature']

    message = "Currently, it's {} degrees outside. There's also {}".format(temperature, summary)
    print(message)


if __name__ == "__main__":
    get_weather(latitude = 37.8267, longitude = -122.4233)

The first step is to launch the virtual environment, install current dependencies, and install the python-dotenv package.

virtualenv env
source env/bin/activate
pip install -r requirements.txt
pip install python-dotenv
pip freeze > requirements.txt

Now that the package is installed, let's create a .env file with the following text.

DARK_SKY_API_KEY=ENTER-YOUR-DARKSKY-API-KEY-HERE

Any variables set in the .env file will be accessible in the app via os. Note that variables set here will override any variables set in your bash_profile. Meaning, if I have one value for my api key in my bash_profile and another set here for the same variable in the .env file, the value in the .env file will be used in the app.

Now that the .env file is set up, I can update the weather_app.py file to look like the following

from dotenv import load_dotenv
import os
import requests

load_dotenv()

DARK_SKY_API_KEY = os.environ['DARK_SKY_API_KEY']

def get_weather(latitude, longitude):
    url = "https://api.darksky.net/forecast/{}/{},{}".format(DARK_SKY_API_KEY, latitude, longitude)

    response = requests.get(url)
    data = response.json()

    current_weather = data['currently']
    summary = current_weather['summary'].lower()
    temperature = current_weather['temperature']

    message = "Currently, it's {} degrees outside. There's also {}.".format(temperature, summary)
    print(message)


if __name__ == "__main__":
    get_weather(latitude = 37.8267, longitude = -122.4233)

Make sure you add your .env file to the project's .gitignore file so that it doesn't get tracked in git history and sent to Github/ Bitbucket/ anywhere else.

That's it! Running python weather_app.py shows that the app works! You can share your code with anyone without having to worry about them using your credentials. Here's what's logged out to my console.

Currently, it's 56.7 degrees outside. There's also overcast.

Link to commit

Cleanup

Although the app works perfectly fine, I like to move all of my environment variable imports to a config.py file. This way, if multiple files need to reference the same value, I don't have to use dotenv across my entire app.

Move the following code to a config.py file.

from dotenv import load_dotenv
import os

load_dotenv()

DARK_SKY_API_KEY = os.environ['DARK_SKY_API_KEY']

Now, the top of the weather_app.py file only needs to contain the following lines of code:

import requests

from config import DARK_SKY_API_KEY

def get_weather(latitude, longitude):
...

The app runs the same, but now it's a little cleaner and easier to expand on in the future.

Link to repo

Athena, Can I get a Wake Up Call?

Taylor Facen — Sun, 07 Jul 2019 22:15:41 +0000

Now that the foundation of my bot is set up, I can add my first (and probably my favorite) feature - daily wake up calls. It usually takes a lot to wake me up in the morning since I'm known for hitting the snooze button at least 10 times. I might be able to solve this by having Athena call me every morning, and in order to turn the alarm off, I'll have to repeat that day's affirmation. Seems like a positive and effective way of starting my day!

Current Tech Stack
A Quick Note on Using Environment Variables in a Python App
Connecting to a Database
Affirmations
Calling Time!
Am I Awake Yet?

Current Tech Stack

Github Repo
Heroku
Python (Flask)
Postgres Database
Twilio

The two components that I'll be adding to Athena is a Postgres database and Twilio integration. I wish this was the part where I explain my deep thought process on why I chose SQL over NoSQL. To be completely honest, I decided to go the Postgres route for three main reasons:

I'm very familiar with SQL
The Flask ecosystem works well with Postgres
I recognize that projects evolve over time. If something comes up in the future that warrants a more thorough investigation, I'll do it then. Sometimes it's better to make a decision and go with it until a change is necessary than to spend days laboring over a choice that might not even be that critical.

Integrating with Twilio was a no-brainer. Their API is easy to work with, and their documentation is extremely understandable, even for someone who doesn't have a lot of coding experience. I also appreciate that there's plenty of examples in various programming languages. I was able to use snippets from the Python examples to create almost all of the code used for the wake up call feature.

Alright, that's enough sales pitching. Let's get started.

If you're like me, you HATE editing your local environment variables in your bash_profile. Not only do I have to look up this process every time, I also have a few projects with conflicting variables (e.g. the database for one project won't be the same database for another). I alleviate this headache by using a .env file that contains all of the keys and values that I'll need for my project (of course, this isn't uploaded to my repo).
.env

DATABASE_URL=INSERT_DATABASE_URL_HERE

I then use the python-dotenv package to load all keys and values from the .env file. After this, I can access any environment variable from this file via os.environ!
config.py

import os
from dotenv import load_dotenv

load_dotenv()

DATABASE_URL = os.environ['DATABASE_URL']

Connecting to a Database

Code Checkpoint

I decided to use two separate databases for my project - one for development and testing and the other one for production. This is because no matter what branch or environment code is sitting in, I want it to mimic as much of the real-life production code as possible. One alternative to this approach would be to have a database schema for development and one for testing. While this might work for small projects, I plan on having many tables grouped into schemas (e.g. tasks, budget, health, etc.).

Adding a database is easy. All I had to do was head over to the "Resources" tab on my application, search for "Postgres", and click "provision". That's it!. Now my app has a config variable called "DATABASE_URL" that can be used to access the database.

Next up, I installed Flask-SQLAlchemy, added my database's URL as a local environment variable, and updated my create_app function.

app.py

from flask import Flask
from flask_restplus import Api
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

from .general import general_api
from config import DATABASE_URL

def create_app():
    app = Flask(__name__)

    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    app.config['SQLALCHEMY_DATABASE_URI'] = DATABASE_URL

    api = Api(
        title = "Athena",
        version = "0.1",
        description = "Athena the personal assistant"
    )

    # Add namespace
    api.add_namespace(general_api, path = '/athena')

    # Initialize app
    api.init_app(app)
    db.init_app(app)

    return app

Let's go ahead and add a test to check to see that the app is connected to the database.

test_db.py

from athena import create_app, db

import pytest

@pytest.fixture
def app():
    app = create_app()
    return app

def test_db_connection(app):
    with app.app_context():
        result = db.session.execute('SELECT 1').fetchone()

    assert result[0] == 1

Alright, let's get this pushed up...oops an error.

One thing to remember about using a Heroku pipeline is that you HAVE to make sure that the environment variables you use locally can also be found in the following locations

Config Vars section in the app's settings
The Config Vars section in the pipeline's settings
the app.json file

After making these changes, everything ran smoothly

Affirmations

Code Checkpoint
Next up, let's get some affirmations in the database for Athena to pull from. I'll add an "affirmations" endpoint with some some CRUD (create, read, update, no delete) operations.

The model for my affirmations table is simple; there's only two columns: affirmation_id, affirmation

from athena import db

class Affirmation(db.Model):
    """Model for affirmations"""
    __tablename__ = 'affirmations'

    affirmation_id = db.Column(db.String, primary_key = True)
    affirmation = db.Column(db.String, nullable = False)

    def __repr__(self):
        return '<Affirmation {}>'.format(self.affirmation)

    def to_dict(self):
        """Converts object to dictionary"""
        return {k: v for k, v in self.__dict__.items() if k[0] != '_'}

Adding the following in my app.py tells the app to create the affirmation table if it doesn't already exist.

with app.app_context():
    from .models import Affirmation
    db.create_all()

Call me paranoid, but I'm always worried that I might accidentally delete a table. So let's just add a test to verify that the affirmations table is there.

test_affirmations.py

from athena import create_app, db

import pytest

@pytest.fixture
def app():
    app = create_app()
    return app

def test_table_exists(app):
    with app.app_context():
        query = '''
        SELECT 
            TABLE_NAME 
        FROM 
            INFORMATION_SCHEMA.TABLES 
        WHERE
            TABLE_SCHEMA = 'public'
        '''
        tables_result = db.session.execute(query).fetchall()
        tables = map(lambda row: row[0], tables_result)

    assert 'affirmations' in tables

Alright, now that that's done, I can add the code to handle any CRUD operations to the affirmations table. After creating the affirmations endpoint and adding the response model, flask-restplus automatically updates the api's documentation.

I went ahead and added tests for each operation as well. After adding in a few affirmations, we can move on to integrating with Twilio.

affirmations = [
    'You are exactly where you need to be.',
    'I’m living in blessings and abundance in every area of my life.',
    'I am gifted and resourceful. I have all the wisdom and power I need to realize my dream.',
    'I appreciate every challenge and difficulty in my life. They are the best times for my growth.',
    'I am born to be happy. Happiness is my life purpose and compass.',
    'All the best opportunities are coming to me for my dream’s realization.',
    'I give myself permission to do what is right for me.',
    'I allow myself to be who I am without judgment.',
    'I give myself the care and attention that I deserve.',
    'My drive and ambition allow me to achieve my goals.',
    'I am always headed in the right direction.',
    'I trust myself to make the right decision.',
    'I put my energy into things that matter to me.',
    'I am learning valuable lessons from myself every day.',
    'I am becoming closer to my true self every day.',
    'I make a difference in the world by simply existing in it.',
    'I am at peace with who I am as a person.',
    "I am the beautiful combination of God's creativity and imagination."
]

Calling Time!

Code Checkpoint

Alright, the moment we've all been waiting for - Athena get's a voice! The workflow of the morning call is:

Every day at a specific time, Heroku Scheduler will tell my app to run a command that calls my phone
This call sends a request to my app with instructions on where to route my call (in this case, a "wake_up_call" task).
The "wake_up_call" task starts off with a Good Morning message, and then sends a request to my app to get instructions on which affirmation to say and how to validate it
Athena waits for me to repeat the affirmation back and then checks to see if it mostly matches the original phrase (wooh NLP!)

Autopilot Set Up
The two Twilio services that will be used for this feature are Programmable Voice and Autopilot. Voice handles normal calling features, and Autopilot is Twilio's machine learning engine that's used to facilitate user <-> bot interactions. Most of the setup can be done within the console. However, I'm going to programmatically implement some of the setup process so that it's reproducible and tracked with version control.

To connect with Twilio, I installed the Twilio package and added the following to my .env file

ACCOUNT_SID=INSERT_TWILIO_ACCOUNT_SID
AUTH_TOKEN=INSERT_TWILIO_AUTH_TOKEN
TWILIO_NUMBER=INSERT_TWILIO_NUMBER
PHONE_NUMBER=INSERT_PERSONAL_NUMBER
ASSISTANT_SID=INSERT_ASSISTANT_SID

Now my config file looks like this.

import os

from dotenv import load_dotenv

load_dotenv()

# Database
DATABASE_URL = os.environ['DATABASE_URL']

# Twilio Credentials
ACCOUNT_SID = os.environ['ACCOUNT_SID']
AUTH_TOKEN = os.environ['AUTH_TOKEN']
TWILIO_NUMBER = os.environ['TWILIO_NUMBER']
PHONE_NUMBER = os.environ['PHONE_NUMBER']
ASSISTANT_SID = os.environ['ASSISTANT_SID']

The ACCOUNT_SID and AUTH_TOKEN came from my account console, I purchased a TWILIO_NUMBER with calling and messaging capabilities, PHONE_NUMBER is my personal number, and I created a Twilio Autopilot Assistant through the console and wrote down its ASSISTANT_SID. Just so that I don't forget, I'll go ahead and add all of my config variables across the two apps, test settings, and my app.json file.

I added a athena_setup folder that contains the code used to set up her stylesheet (e.g. voice style, failure messages, etc) and another to setup the wake up call task. I also added a model_build.py file that generates a new model build for the assistant. One thing I learned in this process (which hopefully will save you a few headaches) is that you HAVE to run a model build after creating a task or making significant changes to a task. If not, the assistant might not be able to parse any messages you say or text to your bot.

After adding the endpoints for the wake up call workflow, I can push up these changes to Twilio.

Wake Up Call Workflow

The overall set up was pretty straight forward. First, I added a command for initiating the call (requires the Flask_Script package).
manage.py

from flask_script import Manager
from twilio.rest import Client

from athena import create_app
from config import ACCOUNT_SID, AUTH_TOKEN, TWILIO_NUMBER, PHONE_NUMBER, API_BASE

app = create_app()
manager = Manager(app)

@manager.command
def initiate_wake_up_call():
    client = Client(ACCOUNT_SID, AUTH_TOKEN)

    call = client.calls.create(
        url = '{}/assistant/route?target_task=wake_up_call'.format(API_BASE),
        to = PHONE_NUMBER,
        from_ = TWILIO_NUMBER
    )

    return call.sid

if __name__ == "__main__":
    manager.run()

Then, I added three API calls to handle different aspects of the wake up call.

/route for routing the call to the correct assistant task

@api.route('/route')
class Router(Resource):
    @api.expect(route_parser, validate = True)
    def post(self):
        args = route_parser.parse_args()
        target_task = args.get('target_task', None)

        response = VoiceResponse()
        connect = Connect()
        connect.autopilot(ASSISTANT_SID, TargetTask = target_task)
        response.append(connect)

        response = make_response(str(response))
        response.headers['Content-type'] = 'text/html; charset=utf-8'

        return response

wake_up_call/dialog for creating the right dialog with a randomly selected affirmation

@api.route('/wake_up_call/dialog')
class WakeUpCallDialog(Resource):
    def post(self):
        client = current_app.test_client()
        resp = client.get('/athena/affirmations?random=True').json

        affirmation = resp['affirmation']
        affirmation_id = resp['affirmation_id']

        return create_wake_up_call_actions(affirmation, affirmation_id)

def create_wake_up_call_actions(affirmation, affirmation_id):
    webhook_url = "{}/assistant/wake_up_call/validate?affirmation_id={}".format(API_BASE, affirmation_id)
    actions = {
        "actions": [
            {
                "collect": {
                    "name": "affirmation_dialog",
                    "questions": [
                        {
                            "question": affirmation,
                            "name": "affirmation_respoinse",
                            "validate": {
                                "webhook": {
                                    "url": webhook_url,
                                    "method": "POST"
                                },
                                "on_success": {
                                    "say": "Wooh! Let's get the day started."
                                }
                            }
                        }
                    ],
                    "on_complete": {
                        "redirect": "task://hello_world"
                    }
                }
            }
        ]
    }

    return actions

/wake_up_call/validate for checking my response against the original affirmation to validate that I said it correctly

@api.route('/wake_up_call/validate')
class WakeUpCallValidate(Resource):
    @api.expect(wake_up_call_validate_parser, validate = True)
    def post(self):
        args = wake_up_call_validate_parser.parse_args()

        affirmation_id = args.get('affirmation_id')
        affirmation_response = args.get('ValidateFieldAnswer')

        # Get affirmation text
        client = current_app.test_client()
        resp = client.get('/athena/affirmations?affirmation_id=' + affirmation_id).json
        affirmation = resp['affirmation']

        return validate_wake_up_call_response(affirmation, affirmation_response)

def validate_wake_up_call_response(affirmation, affirmation_response):
    def transform(text):
        # Lower text
        text_lower = text.lower()
        # Strip out punctuation
        text_alpha_num = text_lower.translate(str.maketrans('', '', string.punctuation))

        return text_alpha_num

    affirmation_response = transform(affirmation_response)
    affirmation = transform(affirmation)

    jaro_distance = jellyfish.jaro_distance(affirmation, affirmation_response)

    if jaro_distance >= 0.75:
        is_valid = True
    else:
        is_valid = False

    return {"valid": is_valid, 'jaro_distance': jaro_distance}

In order to check my response against the original phrase, there's some initial text transformation that has to be applied. First, both texts are converted to lowercase, and then all punctuation is removed. Then, I use the jellyfish package to compute the Jaro distance between both strings. If the distance is greater than or equal to 0.75 (my personal threshold), then I'll count that as a match.

Am I Awake Yet?

Code Checkpoint

Although I have the best intentions for waking up in the morning, sometimes my "sleepy" self acts against these intentions. So to make sure that I just don't pick up the phone and hang up, I'll make a few tweaks so that Athena calls every 15 minutes if the previous call wasn't successful. Doing this was pretty tricky. Although it's easy to get a list of previous calls, it's not possible to check previous dialogs to see if they're successful. My workaround is that when my '/validate' API says that my response is valid, it will give that call a feedback star count of 5. Then, in my command, I'll have the calling code in a while loop that only breaks if the previous call has a feedback star count of 5.

manage.py

@manager.command
def initiate_wake_up_call():
    client = create_twilio_client()
    call_complete = False

    while not call_complete:
        call = client.calls.create(
            url = '{}/assistant/route?target_task=wake_up_call'.format(API_BASE),
            to = PHONE_NUMBER,
            from_ = TWILIO_NUMBER
        )

        call_sid = call.sid

        call_status = 'in-progress'
        while call_status != 'completed':
            time.sleep(30)
            last_call = client.calls(call_sid).fetch()
            call_status = last_call.status

        try:
            feedback_score = last_call.feedback().fetch().quality_score
        except:
            feedback_score = 1

        if feedback_score == 5:
            call_complete = True
        else:
            # Snooze time
            time.sleep(15 * 60)

    return {'status': 'complete'}

wake_up_call.py

def validate_wake_up_call_response(affirmation, affirmation_response):
    client = create_twilio_client()

    def transform(text):
        # Lower text
        text_lower = text.lower()
        # Strip out punctuation
        text_alpha_num = text_lower.translate(str.maketrans('', '', string.punctuation))

        return text_alpha_num

    affirmation_response = transform(affirmation_response)
    affirmation = transform(affirmation)

    jaro_distance = jellyfish.jaro_distance(affirmation, affirmation_response)

    # Get the last call (should be current call)
    last_call = client.calls.list(limit = 1)[0]

    if jaro_distance >= 0.75:
        is_valid = True
        feedback = client.calls(last_call.sid).feedback().create(quality_score = 5)
    else:
        is_valid = False
        feedback = client.calls(last_call.sid).feedback().create(quality_score = 1)

    return {"valid": is_valid, 'jaro_distance': jaro_distance}

Running the task every day

The last thing I need to do is to tell Heroku to run my task every day at a specific time. I'm going to be ambitious here and pick 5:00 AM EST (9:00 AM UTC) as my run time. Heroku makes this process really simple. All I have to do is add the Heroku Scheduler resource to my app and type the following.

python manage.py initiate_wake_up_call

Voila! That's it. After updating Athena's intro and version number and redeploying my task with my production API_BASE, I'm done!

Thanks for reading along! Stay tuned for more walkthroughs.

Athena v0.2

Setting up the Backend for Athena, my Personal Assistant

Taylor Facen — Wed, 03 Jul 2019 00:22:01 +0000

If I'm going to build a full-featured, AI-powered, totally-awesome personal assistant bot, I need to start off with a solid foundation. Here's a post that outlines all of the involved technologies and my initial flask app set up. If you would like to see how the project looked at different stages of the process, click on the Code Checkpoint link to head to the commit in Github.

Tech Stack
Initial Commit
Deployment with Heroku
Testing... 1, 2, 3
Athena, Tell Me More about Yourself

Current Tech Stack

Github Repo
Heroku
Python (Flask)

Initial Commit

Code Checkpoint

Ahh, who doesn't love typing Initial Commit to get a project going? To kick things off, I started with a basic Hello World flask app with Swagger documentation via flask-restplus. Although restplus isn't a requirement for deploying flask app, I find it much simpler to have the service provide the documentation for me. Also, it allows for testing endpoints without having to use curl or an IDE.

Virtualenv
As with any Python project, I start off by adding a virtual environment. This step is crucial in making sure that what I see on my computer is the same thing I'll see when deployed (and so that people like you can copy my code with ease).

Flask Application Factory
Flask suggests to use the application factory for two main reasons:

Easy Testing
Having multiple instances of the same (or slightly tweaked) app

You can read more about this method here.

Folder Hierarchy and App Structure
To be honest, I'm still learning about the best organization method for flask so that projects are easy to follow and update. Here's the structure that I have at this point:

Because Athena will have multiple components (because no one's life only pertains to one area), the api will have multiple namespaces (think folders) that will organize similar endpoints. I'm starting off with a "general" section to house all foundational api calls and activities..

After getting the project set up, this is what the app looks like when deployed locally

python run.py

And heading to /athena/intro in the browser shows the following response

Deployment with Heroku

Code Checkpoint

Alright, now it's time to push this out to the public. I'm using Heroku to host my app for the following reasons:

I can connect a branch to my app so that whenever code is push to remote, a new app with the new changes is deployed
Automated testing
App pipelines
The interface is amazingly user-friendly
When there's a pull request, Heroku can automatically deploy a one-off version of the app with the code in the pr
plus a million more reasons

Pipeline
I went ahead and created a pipeline with two apps for my two development environments:

production: this is what I'll interact with every day
staging: this is where I'll deploy code that's finished with development but may not be 100% ready to go public (e.g. completed sub-features that need another sub-feature to be finished before the entire section is complete)

The entire pipeline is connected to my project repo, and my prod app is connected specifically to my master branch. Once the staging branch is created down the line, I'll go ahead and connect that branch to the staging app (don't want to get ahead of myself).

The prod app is set to automatically deploy with each push. However, before the first deployment, I need to add a few things to the app to make it Heroku-ready.

Changes

Added Procfile (file)
Installed gunicorn (package)
Created app instance in run.py

After adding these changes, I ran heroku local in my terminal to make sure I had everything set up right.

Within seconds of pushing my changes to Github, Heroku started to deploy my changes to the app.

And voila! Athena is public!

Testing... 1, 2, 3

Code Checkpoint

One thing I've learned along my coding journey is that sooo much time can be saved by starting off a project with testing in mind. Heroku and pytest make it easy to make sure that each push is a fully functioning application.

First up, let's add a test to make sure that the app works.

Changes

Installed pytest (package)
Added test folder with test
Enabled CI in pipeline
Added app.json file for test configuration
Added pytest.ini (file)

After getting pytest installed, I added a simple test to check to see if the app is running. One quick way to do this is to check to see if a GET request to '/' returns with a status code of 200.

test_app.py

from athena import create_app

import pytest

@pytest.fixture
def app():
    app = create_app()
    return app

def test_app(app):
    client = app.test_client()

    resp = client.get('/')
    assert resp.status_code == 200

Tip
If you've used pytest with a Python 3.# virtualenv before, you've probably seen the following error:

DeprecationWarning: Using or importing the ABCs from 'collections' instead of from 'collections.abc' is deprecated, and in 3.8 it will stop working
    from collections import Hashable

To make sure that DeprecationWarnings like this don't halt the entire app, you can add a pytest.ini file with the following:

[pytest]
filterwarnings = ignore::DeprecationWarning

After running pytest in the terminal, the following output is displayed.

Next step is to add the app.json file so that Heroku knows how to test the app. Afterwards, I can enable CI on the pipeline and then deploy the changes for testing. Here's what's displayed on the screen after pushing to remote.

And here's what's displayed when all tests pass.

Athena, Tell Me More about Yourself

Code Checkpoint
Okay, the set up is basically done. We have automatic deployment and testing and a solid application organization structure. To finish things out, I'll go ahead and create a staging application for any future features. Also, just to add a little spice to this tutorial, let's try to commit the first feature change so that we can get a glimpse into the power of Heroku's Review Apps.

Changes

Added staging branch
Connected staging branch to staging app with automatic deployment
Added more details to the GET /athena/intro request

After enabling Review Apps (inheriting config vars from the staging application), changing the response to the intro request, and submitting a pull request, an app was created and launched for me to see the changes in real time. The pull request page shows that all tests passed and gives a link to view the deployment (can also see from the heroku dashboard).

Wooh! All tests passed and my intro text was updated. Time to accept the pr, push the changes up to master, and release v0.1 of my app!

Thanks for reading along! Next, I'll program Athena to give me a wake up call every morning.

Athena v0.1

Building a Crypto Trading Bot in Python 101: Automatic Deposits on Coinbase Pro

Taylor Facen — Tue, 30 Apr 2019 01:47:12 +0000

Coinbase Pro is a pretty easy to use crypto exchange. Although it doesn’t allow trading of the entire crypto universe, it does allow users to deposit and trade with fiat currencies. One feature that’s missing, but pretty easy to implement, is automatic deposits.

This tutorial will walk you through using Python to initiate deposits on a periodic basis. I know what you're thinking, "Making automatic deposits does not constitute having a trading bot". Future tutorials will build on this one to complete out all of the bot features. Stay tuned!

Want to skip to the end? Here's the repo with this tutorial's code.

Technologies used

Python - Programming Language
Logger - Python package for logging events
AWS S3 - Storage for logging events
Heroku - Deployment
Coinbase Pro - Crypto Exchange

Coinbase Pro Set Up
Heroku Set Up
Connecting to Coinbase Pro
Making Deposits
App Logging (optional)
Environment Variables in Heroku
Extra notes

Coinbase Pro Set Up

To get started, make sure you have a production account with Coinbase Pro and an account on the Coinbase Pro Sandbox (this is used for testing). For both accounts, create API credentials

Profile in upper right corner -> API -> New API Key
Permissions - Later on, I'll show you how to place trades, so for now you can select all three (view, transfer, trade)
Passphrase - leave as default
IP Whitelist - leave as default
Two Factor Authentication Code - depends on how you set up your account

When you're done, you should have the following for both your production and sandbox accounts:

Passphrase
Secret
Key

Flask App

First create a folder for this project and then initiate and activate a virtual environment

mkdir crypto_trading_bot
cd crypto_trading_bot
virtualenv env
source env/bin/activate

Let's go ahead and install Flask and create the requirements file.
Note: Make sure your virtual environment is activated. If not, all of the packages installed on your computer will be added to the requirements file.

pip install Flask
pip freeze > requirements.txt

Next, create a file with the following
Filename: app.py

from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/", methods = ['GET'] )
def func():
    return jsonify({"message": "Hello World"})

if __name__ == '__main__':
    app.run(debug=True)

This code create a basic Flask app with a single endpoint. Let's test it out. In your terminal write

python app.py

. This runs the Flask app on your local computer. In your terminal, you should see a line that says something like "Running on http://###.#.#.#:###/". If you go to this URL in your browser, you should see "message: Hello World"

Alright, now that the app is set up, let's deploy it to Heroku.

Heroku Set Up

Usually, tutorials walk you through the entire process before showing you how to deploy it. I'm going to go through the deployment step now because Heroku allows for automatic deployments when code is pushed to a branch. Also, from this point on, we'll need to utilize secure information like secrets and passphrases. Using Heroku's environment configuration is way easier than adding all of the keys to my computer's bash_profile, and doing so makes sure you don't mix keys in different projects.

If you haven't already, create an account on Heroku. Then, install the Heroku CLI to your computer and go through the set up process. Afterwords, create an app and head over to the Deploy page to connect it to the repository where you're pushing your code.

Deployment Method - Github (feel free to use Heroku Git if you want)
Automatic deploys - Select the branch you want to automatically deploy and then click "Enable Automatic Deploys"

Before we can check out our app, we have to serve it. We'll do so using Gunicorn. Run the following in your terminal (make sure your virtualenv is still running).

pip install gunicorn
pip freeze > requirements.txt

Then, add the following file.
Filename: Procfile
(warning, there's no extension to this file name)

web: gunicorn app:app

Now commit your changes, wait about a minute, and then head over to your site! You can find your domain on the "Settings" page in the "Domain and certificates" section.

Now that that's set up, let's connect the app to Coinbase Pro.

Connecting to Coinbase Pro

Let's create an environment file so that we can deploy the app locally for testing.

Filename: .env

CB_KEY=ENTER_YOUR_SANDBOX_KEY_HERE
CB_PASSPHRASE=ENTER_YOUR_SANDBOX_PASSPHRASE_HERE
CB_SECRET=ENTER_YOUR_SANDBOX_SECRET_HERE
CB_URL=http://api-public.sandbox.pro.coinbase.com

If you're uploading your code to a public repo, make sure to add .env to your .gitignore file.

Next, let's install the Coinbase Pro Python package.

pip install cbpro
pip freeze > requirements.txt

Because I already know that I'll be adding multiple functions (modules) that will need to connect to my Coinbase Pro account (making deposits, placing trades, viewing balances, etc.), I'm going to create a decorator function that will pass my Coinbase authenticated client to any necessary functions. Don't worry, I actually JUST learned about decorator functions. So if this part doesn't make sense to you, don't worry. Decorators don't make sense to most of us.

Filename: cbpro_client.py

import cbpro

from config import CB_CREDENTIALS

def get_client(credentials):
    """Returns the cbpro AuthenticatedClient using the credentials from the parameters dict"""

    cbpro_client = cbpro.AuthenticatedClient(credentials['KEY'], credentials['SECRET'], credentials['PASSPHRASE'], api_url=credentials['URL'])
    return cbpro_client

def cbpro_client(func):
    def function_wrapper(*args, **kwargs):
        cbpro_client = get_client(CB_CREDENTIALS)
        resp = func(cbpro_client = cbpro_client, *args, **kwargs)

        return resp

    return function_wrapper

In the same folder, create a config file that will read your credentials from the .env file (when using Heroku locally) or your bash_profile (when using python app.py).

Filename: config.py

# Coinbase Credentials
CB_CREDENTIALS = {
    'PASSPHRASE': os.environ['CB_PASSPHRASE'],
    'SECRET': os.environ['CB_SECRET'],
    'KEY': os.environ['CB_KEY'],
    'URL': os.environ['CB_URL']
}

Next up, we're making a deposit!

Making Deposits

Alright, this is the moment we've all been waiting for.

Filename: deposit_funds.py

from cbpro_client import cbpro_client

@cbpro_client
def get_deposit_account(cbpro_client):
    """ Gets ID of account's ACH bank account (assumes there's only one)

    Params:
        - None
    Return:
        - account: dict
        {
            {
            "allow_buy": bool, 
            "allow_deposit": bool, 
            "allow_sell": bool, 
            "allow_withdraw": bool, 
            "cdv_status": str, 
            "created_at": time, 
            "currency": str, 
            "id": str, 
            "limits": dict
            "name": str, 
            "primary_buy": bool, 
            "primary_sell": bool, 
            "resource": str, 
            "resource_path": str, 
            "type": str, 
            "updated_at": time, 
            "verification_method": str, 
            "verified": bool
            }
        }
    """

    bank_accounts = cbpro_client.get_payment_methods()

    for account in bank_accounts:
        # This assumes that there is only one ACH bank account connected
        if account['type'] == 'ach_bank_account':
            return account



@cbpro_client
def deposit_funds(cbpro_client, deposit_amount = 10): # Default deposit amount is $10
    """ Makes deposit into USD Wallet

    Params: 
        - deposit_amonut: int (default 10)
    Return: 
        - deposit response
        {
            'id' : str,
            'amount' : str,
            'currency' : 'USD',
            'payout_at' : str (datetime)
        }
    """

    deposit_account_id = get_deposit_account()['id']

    resp = cbpro_client.deposit(deposit_amount, 'USD', deposit_account_id)

    return resp

Let's unpack. The first function retrieves the bank account id for the account used to make the deposit. This function assumes that there's only one ACH account. So, if you have multiple ach accounts connected to your Coinbase account, feel free to customize how the account is selected. The second function uses the returned bank account to make a deposit.

*Random: I'm still trying to figure out my docstring style. If anyone has any tips or suggestions, feel free to share!

Now, let's add this function to our app so that we can test it out! In your app.py file, add the following to the top

from deposit_funds import deposit_funds

And right above "if name == 'main':", add a new endpoint

@app.route("/deposit", methods = ['GET'])
def deposit_function():
    resp = deposit_funds()
    return jsonify(resp)

Normally, this is where I would say, "To test it out, run the following in your terminal and then head to the deposit endpoint (http://0.0.0.0:####/deposit)"

heroku local

However, this would produce an error because Coinbase doesn't let you deposit funds via API in sandbox mode. So unfortunately, this won't work. However, when you switch over to your real Coinbase credentials, it will be good to go! If you don't believe me (it's okay, we just met), then import the get_deposit_account function instead and use that function for the deposit endpoint. After running your app locally, you should see a JSON object of your deposit account in the browser. Don't forget to switch back to the deposit_funds function.

Now I don't know about you, but I think it would be pretty nice to see the output of each step within the code so I can make sure everything is up and running correctly. Next, we'll add some logging.

Logging

*Feel free to skip this step if logging isn't your flavor

Logging is an AWESOME tool to help you see what's going on in your code (no more print statements!). Because I want to be extra fancy, we'll publish log files to AWS S3 buckets so that we can view them easily.

Here's a tutorial on creating an IAM user on AWS. Also, here's how you create a bucket on AWS S3. Create a bucket and write down the name (e.g. crypto_bot_logs)

Let's create our logger.
First, install boto3, a package to access AWS resources via Python

pip install boto3
pip freeze > requirements.txt

Second, add your IAM credentials to your .env file
Filename: .env

AWS_ACCESS_KEY_ID=ENTER_YOUR_KEY_HERE
AWS_SECRET_ACCESS_KEY=ENTER_YOUR_SECRET_HERE

Thirdly, grab your credentials in your config file by adding the following to the end
Filename: config.py

# AWS Credentials
AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID']
AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY']

Fourthly (is fourthly even a word?), create your logger file
Filename: logger.py

import boto3
from datetime import datetime
import logging

from config import AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY

s3_client = boto3.client('s3', 
        aws_access_key_id = AWS_ACCESS_KEY_ID, 
        aws_secret_access_key = AWS_SECRET_ACCESS_KEY,
        region_name = 'us-east-1')

def create_logger(logger_name):
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)
    # create file handler which logs even debug messages
    fh = logging.FileHandler('{}_{}.log'.format(logger_name, datetime.now().isoformat().split('.')[0]))
    fh.setLevel(logging.DEBUG)
    # create formatter and add it to the handlers
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    fh.setFormatter(formatter)
    # add the handlers to the logger
    logger.addHandler(fh)

    return logger

def upload_log(logger):
    file_name = logger.handlers[0].baseFilename
    directory = datetime.now().date().isoformat()
    key = "{}/{}".format(directory, file_name.split('/')[-1])
    bucket_name = ENTER_YOUR_BUCKET_FILENAME_HERE

    s3_client.upload_file(Filename = file_name, Bucket = bucket_name, Key = key)

def logger(func):
    def function_wrapper(*args, **kwargs):
        function_name = func.__name__
        logger = create_logger(function_name)
        logger.info("Now running - {}".format(function_name))

        resp = func(logger = logger, *args, **kwargs)
        upload_log(logger)

        return resp

    return function_wrapper

Here's what each function does:

create_logger - creates a logger with a FileHandler
upload_log - Uploads the recently created log file to the designated file
logger - This is a decorator function that creates a logger, passes it to the function, and then uploads the log file when the function finishes

Now, we can add some logging statements to our functions. Here's what I added, but feel free to customize based on what you want to see. Here's an awesome tutorial on logging.

Here's how my deposit_funds function looks now. Don't forget to import logger at the top

Filename: deposit_funds.py

@cbpro_client
@logger
def deposit_funds(cbpro_client, logger, deposit_amount = 10): # Default deposit amount is $10
    """ Makes deposit into USD Wallet

    Params: 
        - deposit_amonut: int (default 10)
    Return: 
        - deposit response
        {
            'id' : str,
            'amount' : str,
            'currency' : 'USD',
            'payout_at' : str (datetime)
        }
    """
    logger.info("Getting account ID")
    deposit_account_id = get_deposit_account()['id']
    logger.info("Account ID: {}".format(deposit_account_id))

    resp = cbpro_client.deposit(deposit_amount, 'USD', deposit_account_id)
    if 'message' in resp.keys():
        logger.warning("In sandbox mode, unable to make deposit")
    else:
        logger.info("Deposit Response: {}".format(resp))

    return resp

If you run your app now (heroku local) and head to the '/deposits' endpoint, you'll see the same error you saw in the previous section. However, if you head to your S3 bucket, open the folder with today's date, and open the file named "deposit_funds_[todays_date].log", you should see something like this.

Yay! Now we can see where in the function the error is coming from. When you deploy the app to Heroku and use your production credentials, the log message will show you the response for making the deposit.

That's (almost) it! Commit your changes and push to your repo so it can make its way to Heroku.

Environment Variables in Heroku

Remember how we created a .env file so that we could read environment variables when using heroku local? Well, now that it's game time, it's time to add our real credentials to our Heroku app. On your app's Settings page in Heroku, click on "Reveal Config Vars" and add the following key value pairs.

key: CB_KEY value: YOUR_PRODUCTION_CB_KEY
key: CB_PASSPHRASE value: YOUR_PRODUCTION_PASSPHRASE
key: CB_SECRET value: YOUR_PRODUCTION_SECRET
key: CB_URL value: https://api.pro.coinbase.com
key: AWS_ACCESS_KEY_ID value: YOUR_AWS_ACCESS_KEY
key: AWS_SECRET_ACCESS_KEY value: YOUR_AWS_SECRET_KEY
key: TZ value: Your Timezone (e.g. America/New_York) (optional)

Now, when your app runs, it will use these credentials.

Schedule

If you've made it this far, congrats! You now have an app that makes a deposit of $10 whenever you go to http://your_domain_url.com/deposit. But what if you want the app to do it automatically? Next, I'll show you how to make some slight changes so that the deposit_funds function runs on a schedule.

Alright, one more package to install. In your terminal run the following:

pip install flask_script
pip freeze > requirements.txt

Then create the following file.
Filename: manage.py

from flask_script import Manager

from app import app
from deposit_funds import deposit_funds

manager = Manager(app)

@manager.command
def make_deposit():
    deposit_funds()

This file creates a command that we can call from Heroku.

Then from the Resources tab on your Heroku dashboard, search for Heroku Scheduler and add it to your app. Note: Heroku might ask you to add payment information to your account. Click on "Heroku Scheduler" and then "Create job". I want my function to run every Thursday. Here's how my job looks

Run Command: if [ "$(date +%u)" = 4 ]; then python manage.py make_deposit; fi

Click "Save Job" to lock it in. You're done! You now have an app that will make automatic deposits to your Coinbase Pro account and log activities to a S3 bucket!

Extra notes

Another alternative is to create an AWS Lambda function and that gets called via AWS Cloudwatch. If you're interested in seeing how to do it this way, let me know!

Forem: Taylor Facen

Finta's Automated Financial Synchronization powered by Plaid, Stripe and Inngest

Problem Overview

DRY Functions

Function Orchestration

Key Inngest Features That We Use

concurrency with a key

invoke

Implementation Feature 1: Composability (DRY Programming)

Implementation Feature 2: Function Orchestration

Takeaways for you to use in your Inngest Project

I Built My Own Verzuz App

What I built

Category Submission: Exciting X-Factors

Demo Link

Link to Code

How I built it (what's the stack? did I run into issues or discover something new along the way?)

Additional Resources/Info

How I Used Twilio Autopilot to Build an Interview Coach

Development

Lessons Learned

What's Next?

How to Safely Add Environment Variables to your Python App

What we’re Building

Let’s Code!

Cleanup

Athena, Can I get a Wake Up Call?

Contents

Current Tech Stack

Connecting to a Database

Affirmations

Calling Time!

Am I Awake Yet?

Running the task every day

Setting up the Backend for Athena, my Personal Assistant

Contents

Current Tech Stack

Initial Commit

Deployment with Heroku

Testing... 1, 2, 3

Athena, Tell Me More about Yourself

Building a Crypto Trading Bot in Python 101: Automatic Deposits on Coinbase Pro

Contents

Coinbase Pro Set Up

Flask App

Heroku Set Up

Connecting to Coinbase Pro

Making Deposits

Logging

Environment Variables in Heroku

Schedule

Extra notes

`concurrency` with a key

`invoke`