💡Never forget again: Build a scheduled reminder app in <50 lines of Python using REST and Postgres

🚀 Ever forget something important? Let's build a reliable, scheduled reminder app in under 50 lines of Python.

In this tutorial, you'll create and deploy a crashproof Python backend that schedules reminder emails for any date in the future. You can try the live app here: simply enter your email address and a date, and you’ll receive a reminder on that date!

TL;DR

This application consists of:

• A single Python backend service
• Multiple REST APIs
• A Postgres database

We'll be using:

• FastAPI to define REST APIs and handle HTTP requests.
• SendGrid to send emails.
• DBOS to make the backend durable and serverlessly host it in the cloud.

All source code is available on GitHub.

---

Import and Initialize the App

Let's start off with imports and initializing the DBOS and FastAPI apps.

import os

from dbos import DBOS, SetWorkflowID
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, EmailStr
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail

app = FastAPI()
DBOS(fastapi=app)

Scheduling Emails

Next, we'll write the workflow for sending emails. First, it will send a quick confirmation email, then wait until the scheduled day to send the reminder.

Thanks to DBOS's durably executed workflow, waiting until the scheduled day is simple—no matter how far away that day is, just use sleep!

Under the hood, when you call DBOS.sleep, it saves the wake-up time in the database. This ensures that even if your program is interrupted or restarted during a days-long sleep, it will still wake up on time to send the reminder.

If you need to schedule recurring events rather than a one-time email, consider using scheduled workflows.

@DBOS.workflow()
def reminder_workflow(to_email: str, send_date: datetime, start_date: datetime):
    send_email(
        to_email,
        subject="DBOS Reminder Confirmation",
        message=f"Thank you for signing up for DBOS reminders! You will receive a reminder on {send_date}.",
    )
    days_to_wait = (send_date - start_date).days
    seconds_to_wait = days_to_wait * 24 * 60 * 60
    DBOS.sleep(seconds_to_wait)
    send_email(
        to_email,
        subject="DBOS Reminder",
        message=f"This is a reminder from DBOS! You requested this reminder on {start_date}.",
    )

Sending Emails

Next, let's write the email-sending code using SendGrid. This will require a few environment variables:

api_key = os.environ.get("SENDGRID_API_KEY", None)
if api_key is None:
    raise Exception("Error: SENDGRID_API_KEY is not set")

from_email = os.environ.get("SENDGRID_FROM_EMAIL", None)
if from_email is None:
    raise Exception("Error: SENDGRID_FROM_EMAIL is not set")

Next, we implement the send_email function using SendGrid's Python API. We’ll annotate this function with @DBOS.step so that the reminder workflow calls it durably and doesn’t re-execute it if restarted.

@DBOS.step()
def send_email(to_email: str, subject: str, message: str):
    message = Mail(
        from_email=from_email, to_emails=to_email, subject=subject, html_content=message
    )
    email_client = SendGridAPIClient(api_key)
    email_client.send(message)
    DBOS.logger.info(f"Email sent to {to_email}")

Serving the App

Next, let’s use FastAPI to create an HTTP endpoint for scheduling reminder emails. This endpoint will accept an email address and a scheduled date, then start a reminder workflow in the background.

As a basic anti-spam measure, we’ll use the provided email address and date as an idempotency key. This ensures that only one reminder can be sent to any given email address per day.

class RequestSchema(BaseModel):
    email: EmailStr
    date: str


@app.post("/email")
def email_endpoint(request: RequestSchema):
    send_date = datetime.strptime(request.date, "%Y-%m-%d").date()
    today_date = datetime.now().date()
    with SetWorkflowID(f"{request.email}-{request.date}"):
        DBOS.start_workflow(reminder_workflow, request.email, send_date, today_date)

Finally, let's serve the app's frontend from an HTML file using FastAPI. In production, we recommend using DBOS primarily for the backend, with your frontend deployed elsewhere.

@app.get("/")
def frontend():
    with open(os.path.join("html", "app.html")) as file:
        html = file.read()
    return HTMLResponse(html)

Try it Yourself!

Setting Up SendGrid

This app uses SendGrid to send reminder emails. Create a SendGrid account, verify an email for sending, and generate an API key. Then, set the API key and sender email as environment variables:

export SENDGRID_API_KEY=<your key>
export SENDGRID_FROM_EMAIL=<your email>

Deploying to the Cloud

To deploy this app to DBOS Cloud, first install the DBOS Cloud CLI (requires Node):

npm i -g @dbos-inc/dbos-cloud

Then clone the dbos-demo-apps repository and deploy:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/scheduled-reminders
dbos-cloud app deploy

This command outputs a URL—visit it to schedule a reminder!
You can also visit the DBOS Cloud Console to see your app's status and logs.

Running Locally

First, clone and enter the dbos-demo-apps repository:

git clone https://github.com/dbos-inc/dbos-demo-apps.git
cd python/scheduled-reminders

Then create a virtual environment:

python3 -m venv .venv
source .venv/bin/activate

DBOS requires a Postgres database.
If you don't already have one, you can start one with Docker:

export PGPASSWORD=dbos
python3 start_postgres_docker.py

Then run the app in the virtual environment:

pip install -r requirements.txt
dbos migrate
dbos start

Visit http://localhost:8000 to schedule a reminder!

Next Steps

Check out how DBOS can make your applications more scalable and resilient:

• Use durable execution to write crashproof workflows.
• Use queues to gracefully manage API rate limits.
• Use scheduled workflows to run your functions at recurring intervals.
• Want to learn what you can build with DBOS? Explore other example applications.

Give it a try and let us know what you think 😊

Comments

[Nikol]

Is DBOS then doing some multithreading in the background of the python app?
Or when DBOS.sleep(seconds_to_wait) saves the continuation time to DBOS is there no more processing happening in the program itself, with DBOS then monitoring and triggering the continuation in the background when the reminder time is reached?

[Qian Li]

Great question! When you call DBOS.sleep(), it saves the wake-up time in the database and then uses time.sleep() to idle the current thread. This database record ensures that if your program is interrupted or restarted during a long sleep, it will still wake up at the exact scheduled time to send the reminder.

[Nikol]

Thanks for the clarification!
Do we then need to manage the threads ourselves, or does @DBOS.workflow() automatically make a new thread for the workflow?

[Qian Li]

@DBOS.workflow() doesn't start a new thread. However, you could use DBOS Queues to start parallel/concurrent tasks (workflows, steps, transactions).