Learn DBOS Python
This tutorial shows you how to use DBOS durable execution to make your Python app resilient to any failure. First, without using DBOS, we'll build an app that records greetings to two different systems: Postgres and an online guestbook. Then, we'll add DBOS durable execution to the app in just four lines of code. Thanks to durable execution, the app will always write to both systems consistently, even if it is interrupted or restarted at any point.
This guide assumes you have a Postgres database running locally. If not, see the quickstart for instructions on how to set it up.
1. Setting Up Your App
Create a folder for your app with a virtual environment, then enter the folder and activate the virtual environment.
- macOS or Linux
- Windows (PowerShell)
- Windows (cmd)
python3 -m venv greeting-guestbook/.venv
cd greeting-guestbook
source .venv/bin/activate
python3 -m venv greeting-guestbook/.venv
cd greeting-guestbook
.venv\Scripts\activate.ps1
python3 -m venv greeting-guestbook/.venv
cd greeting-guestbook
.venv\Scripts\activate.bat
Then, install and initialize DBOS:
pip install dbos
dbos init
dbos migrate
Next, let's use FastAPI to write a simple app that greets our friends. Every time the app receives a greeting, it performs two steps:
- Sign an online guestbook with the greeting.
- Record the greeting in the database.
We deliberately won't use DBOS yet so we can show you how easy it is to add later.
import logging
import os
import requests
from fastapi import FastAPI
from sqlalchemy import create_engine
from .schema import dbos_hello
app = FastAPI()
logging.basicConfig(level=logging.INFO)
# Sign the guestbook using an HTTP POST request
def sign_guestbook(name: str):
requests.post(
"https://demo-guest-book.cloud.dbos.dev/record_greeting",
headers={"Content-Type": "application/json"},
json={"name": name},
)
logging.info(f">>> STEP 1: Signed the guestbook for {name}")
# Create a SQLAlchemy engine. Adjust this connection string for your database.
engine = create_engine(f"postgresql+psycopg://postgres:{os.environ['PGPASSWORD']}@localhost/greeting_guestbook")
# Record the greeting in the database using SQLAlchemy
def insert_greeting(name: str) -> str:
with engine.begin() as sql_session:
query = dbos_hello.insert().values(name=name)
sql_session.execute(query)
logging.info(f">>> STEP 2: Greeting to {name} recorded in the database!")
@app.get("/greeting/{name}")
def greeting_endpoint(name: str):
sign_guestbook(name)
insert_greeting(name)
return f"Thank you for being awesome, {name}!"
Start your app with dbos start
.
To see that it's is working, visit this URL: http://localhost:8000/greeting/Mike
"Thank you for being awesome, Mike!"
Each time you visit, your app should log first that it has recorded your greeting in the guestbook, then that it has recorded your greeting in the database.
INFO:root:>>> STEP 1: Signed the guestbook for Mike
INFO:root:>>> STEP 2: Greeting to Mike recorded in the database!
Now, this app has a problem: if it is interrupted after signing the guestbook, but before recording the greeting in the database, then the greeting, though sent, will never be recorded. This is bad in many real-world situations, for example if a program fails to record making or receiving a payment. To fix this problem, we'll use DBOS durable execution.
2. Durable Execution with Workflows
Next, we want to durably execute our application: guarantee that it inserts exactly one database record per guestbook signature, even if interrupted or restarted.
DBOS makes this easy with workflows.
We can add durable execution to our app with just four lines of code and an import statement.
Copy the following into your main.py
.
import logging
import os
import requests
from dbos import DBOS
from fastapi import FastAPI
from sqlalchemy import create_engine
from .schema import dbos_hello
app = FastAPI()
DBOS(fastapi=app)
logging.basicConfig(level=logging.INFO)
# Sign the guestbook using an HTTP POST request
@DBOS.step()
def sign_guestbook(name: str):
requests.post(
"https://demo-guest-book.cloud.dbos.dev/record_greeting",
headers={"Content-Type": "application/json"},
json={"name": name},
)
logging.info(f">>> STEP 1: Signed the guestbook for {name}")
# Create a SQLAlchemy engine. Adjust this connection string for your database.
engine = create_engine(f"postgresql+psycopg://postgres:{os.environ['PGPASSWORD']}@localhost/greeting_guestbook")
# Record the greeting in the database using SQLAlchemy
@DBOS.step()
def insert_greeting(name: str) -> str:
with engine.begin() as sql_session:
query = dbos_hello.insert().values(name=name)
sql_session.execute(query)
logging.info(f">>> STEP 2: Greeting to {name} recorded in the database!")
@app.get("/greeting/{name}")
@DBOS.workflow()
def greeting_endpoint(name: str):
sign_guestbook(name)
for _ in range(5):
logging.info("Press Control + C to stop the app...")
DBOS.sleep(1)
insert_greeting(name)
return f"Thank you for being awesome, {name}!"
Only the four highlighted lines of code are needed to enable durable execution.
- First, we initialize DBOS on line 12.
- Then, we annotate
sign_guestbook
andinsert_greeting
as workflow steps on lines 17 and 30. - Finally, we annotate
greeting_endpoint
as a durable workflow on line 38.
Because greeting_endpoint
is now a durably executed workflow, if it's ever interrupted, it automatically resumes from the last completed step.
To help demonstrate this, we also add a sleep so you can interrupt your app midway through the workflow.
To see the power of durable execution, restart your app with dbos start
.
Then, visit this URL: http://localhost:8000/greeting/Mike.
In your terminal, you should see an output like:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:root:>>> STEP 1: Signed the guestbook for Mike
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
Now, press CTRL+C stop your app. Then, run dbos start
to restart it. You should see an output like:
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
INFO:root:Press Control + C to stop the app...
INFO:root:>>> STEP 2: Greeting to Mike recorded in the database!
Without durable execution—if you remove the four highlighted lines—your app would restart with a "clean slate" and completely forget about your interrupted workflow. By contrast, DBOS automatically resumes your workflow from where it left off and correctly completes it by recording the greeting to the database without re-signing the guestbook. This is an incredibly powerful guarantee that helps you build complex, reliable applications without worrying about error handling or interruptions.
3. Optimizing Database Operations
For workflow steps that access the database, like insert_greeting
in the example, DBOS provides powerful optimizations.
To see this in action, replace your insert_greeting
with the following:
@DBOS.transaction()
def insert_greeting(name: str) -> str:
query = dbos_hello.insert().values(name=name)
DBOS.sql_session.execute(query)
logging.info(f">>> STEP 2: Greeting to {name} recorded in the database!")
@DBOS.transaction()
is a special annotation for workflow steps that access the database.
It executes your function in a single database transaction.
We recommend using transactions because:
- They give you access to a pre-configured database client (
DBOS.sql_session
), which is more convenient than connecting to the database yourself. You no longer need to configure a SQLAlchemy engine! - Under the hood, transactions are highly optimized because DBOS can update its record of your program's execution inside your transaction. For more info, see our "how workflows work" explainer.
Now, restart your app with dbos start
and visit its URL again: http://localhost:8000/greeting/Mike.
The app should durably execute your workflow the same as before!
The code for this guide is available on GitHub.
Next, to learn how to build more complex applications, check out our Python tutorials and example apps.