Build & Deploy a Fitness App that sends daily E-mails

In this tutorial we build and deploy a fitness app that sends you new workouts per email every day using Python and HarperDB.

In the end, we will have a deployed cloud database and a deployed function that runs every day and sends you a new workout! And all of those services can be used for free!

Tech Stack we use:

• Streamlit for the Frontend
• HarperDB for the Backend
• HarperDB Custom Functions to send emails

Create a new project and install these three dependencies:

pip install harperdb youtube-dl streamlit

Each of these packages is explained in the next steps.

For the Backend we use HarperDB. HarperDB is a distributed database with hybrid SQL and NoSQL functionality and a REST API. The performance benchmarks look very promising and the Cloud instance can be set up in no time.

First, you need to create a HarperDB account and create a cloud instance. No worries, because everything that we build in this project is included in the free tier. After spinning up the cloud instance, create schema named workout_repo, and two tables named workouts and workout_today.

Next, create a file database_service.py and create the code to work with the database and perform basic CRUD operations. The harperdb Python package makes it pretty easy to work with the database:

import harperdb

url = "https://your-clouddatabaseurl.harperdbcloud.com"
username = "YOUR_USER"
password = "YOUR_PW"

db = harperdb.HarperDB(
    url=url,
    username=username,
    password=password
)

SCHEMA = "workout_repo"
TABLE = "workouts"
TABLE_TODAY = "workout_today"


def insert_workout(workout_data):
    return db.insert(SCHEMA, TABLE, [workout_data])

def delete_workout(workout_id):
    return db.delete(SCHEMA, TABLE, [workout_id])

def get_all_workouts():
    try:
        return db.sql(f"select video_id,channel,title,duration from {SCHEMA}.{TABLE}")
    except harperdb.exceptions.HarperDBError:
        return []

def get_workout_today():
    return db.sql(f"select * from {SCHEMA}.{TABLE_TODAY} where id = 0")

def update_workout_today(workout_data, insert=False):
    workout_data['id'] = 0
    if insert:
        return db.insert(SCHEMA, TABLE_TODAY, [workout_data])
    return db.update(SCHEMA, TABLE_TODAY, [workout_data])

(Note: Make sure to replace the credentials in the code with your own username and password.)

As you can see, we leveraged both SQL and NoSQL queries for these operations.

To extract the information from YouTube videos, we use the youtube-dl package. It provides a simple Python API and a command-line tool to download videos from YouTube.com and other video sites.

In our case we don't need to download the videos, but only extract information like title, channel, video_id, duration, and more.

Create a file yt_extractor.py and extract all relevant keys from a YouTube video link:

import youtube_dl
from youtube_dl.utils import DownloadError

ydl = youtube_dl.YoutubeDL()


def get_info(url):
    with ydl:
        try:
            result = ydl.extract_info(
                url,
                download=False
            )
        except DownloadError:
            return None
        
    if "entries" in result:
        video = result["entries"][0]
    else:
        video = result
        
    infos = ['id', 'title', 'channel', 'view_count', 'like_count',
             'channel_id', 'duration', 'categories', 'tags']
    
    def key_name(key):
        if key == "id":
            return "video_id"
        return key
    
    return {key_name(key): video[key] for key in infos}

For the frontend part of our app we use Streamlit, an open-source app framework that lets you turn Python scripts into sharable web apps pretty quickly.

Create a file app.py and put everything together:

import random
import streamlit as st
from yt_extractor import get_info
import database_service as dbs


@st.cache(allow_output_mutation=True)
def get_workouts():
    return dbs.get_all_workouts()

def get_duration_text(duration_s):
    seconds = duration_s % 60
    minutes = int((duration_s / 60) % 60)
    hours = int((duration_s / (60*60)) % 24)
    text = ''
    if hours > 0:
        text += f'{hours:02d}:{minutes:02d}:{seconds:02d}'
    else:
        text += f'{minutes:02d}:{seconds:02d}'
    return text

st.title("Workout APP")

menu_options = ("Today's workout", "All workouts", "Add workout")
selection = st.sidebar.selectbox("Menu", menu_options)

if selection == "All workouts":
    st.markdown(f"## All workouts")
    
    workouts = get_workouts()
    for wo in workouts:
        url = "https://youtu.be/" + wo["video_id"]
        st.text(wo['title'])
        st.text(f"{wo['channel']} - {get_duration_text(wo['duration'])}")
        
        ok = st.button('Delete workout', key=wo["video_id"])
        if ok:
            dbs.delete_workout(wo["video_id"])
            st.legacy_caching.clear_cache()
            st.experimental_rerun()
            
        st.video(url)
    else:
        st.text("No workouts in Database!")
elif selection == "Add workout":
    st.markdown(f"## Add workout")
    
    url = st.text_input('Please enter the video url')
    if url:
        workout_data = get_info(url)
        if workout_data is None:
            st.text("Could not find video")
        else:
            st.text(workout_data['title'])
            st.text(workout_data['channel'])
            st.video(url)
            if st.button("Add workout"):
                dbs.insert_workout(workout_data)
                st.text("Added workout!")
                st.legacy_caching.clear_cache()
else:
    st.markdown(f"## Today's workout")
    
    workouts = get_workouts()
    if not workouts:
        st.text("No workouts in Database!")
    else:
        wo = dbs.get_workout_today()
        
        if not wo:
            # not yet defined
            workouts = get_workouts()
            n = len(workouts)
            idx = random.randint(0, n-1)
            wo = workouts[idx]
            dbs.update_workout_today(wo, insert=True)
        else:
            # first item in list
            wo = wo[0]
        
        if st.button("Choose another workout"):
            workouts = get_workouts()
            n = len(workouts)
            if n > 1:
                idx = random.randint(0, n-1)
                wo_new = workouts[idx]
                while wo_new['video_id'] == wo['video_id']:
                    idx = random.randint(0, n-1)
                    wo_new = workouts[idx]
                wo = wo_new
                dbs.update_workout_today(wo)
        
        url = "https://youtu.be/" + wo["video_id"]
        st.text(wo['title'])
        st.text(f"{wo['channel']} - {get_duration_text(wo['duration'])}")
        st.video(url)

Now we can run the app with the following command:

streamlit run app.py

This will start the server and we can interact with the app. We can now add YouTube videos to our database.

To send emails, we use HarperDB Custom Functions. They allow us to

• Add our own API endpoints to a standalone API server inside HarperDB
• Use HarperDB Core methods to interact with our database
• Run any JavaScript code we want, powered by Fastify

To implement our own custom functions, we need to create a local HarperDB instance as well. For this, follow the instructions in the video starting at 31:00.

Clone the following repo that provides the custom functions code to send an email. Make sure to update helpers/config.js with your database and email credentials, and modify helpers/mail.js to use your name in the email text.

Deploy the custom function in the HarperDB Studio (see the video for more instructions). Now we have an API endpoint that updates the workout of the day with a randomly selected new workout, and then sends an email to the specified address.

The last thing to do is to set up a Cronjob that runs every day and calls the custom functions endpoint. For this, I use the free service https://console.cron-job.org/.

Select POST method, enter your custom functions endpoint url, specify application/json as Content-Type, and put your credentials in the authentication form. In order to not run into an error, also send some dummy data in the request body: "dummy: 1".

Congratulations 🎉!!! Now you should have a working app with a deployed cloud database, and a deployed function that runs every day and sends you a new workout!

The Streamlit app can be used locally to add new workouts to the database whenever needed.

I hope you enjoyed this project! The whole code is also available on GitHub.