DBOS Class

The DBOS class is a singleton—you must instantiate it (by calling its constructor) exactly once in a program's lifetime. Here, we document its constructor and lifecycle methods. Decorators are documented here and context methods and variables here.

class dbos.DBOS

DBOS(
    *,
    fastapi: Optional[FastAPI] = None,
    flask: Optional[Flask] = None,
    config: Optional[ConfigFile] = None,
)

Parameters:

• fastapi: If your application is using FastAPI, the FastAPI object. If this is passed in, DBOS automatically calls dbos.launch when FastAPI is fully initialized. DBOS also adds to all routes a middleware that enables tracing through FastAPI HTTP endpoints.
• flask: If your application is using Flask, the flask object. If this is passed in, DBOS adds to all routes a middleware that enables tracing through Flask HTTP endpoints.
• config: A configuration object. By default, DBOS reads configuration from dbos-config.yaml, but if this object is passed its contents are used instead. We recommend using this for testing only.

launch

DBOS.launch()

Launch DBOS, initializing database connections and starting scheduled workflows. Should be called after all decorators run. You should not call a DBOS function until after DBOS is launched. If a FastAPI app is passed into the DBOS constructor, launch is called automatically during FastAPI setup.

DBOS.launch() connects your app to a Postgres database. It looks for database connection parameters in your dbos-config.yaml and .dbos/db_connection files. If those parameters are set to default values and no database is found, it prompts you to launch a local Postgres database using Docker. If Docker is not found, it prompts you to connect to a database hosted on DBOS Cloud.

Example:

from dbos import DBOS

# Initialize the DBOS object
DBOS()

# Define a scheduled workflow
@DBOS.scheduled("* * * * *")
@DBOS.workflow()
def run_every_minute(scheduled_time: datetime, actual_time: datetime):
    DBOS.logger.info("This is a scheduled workflow!")

# After all decorators run, launch DBOS
DBOS.launch()

Example using Flask:

from flask import Flask
from dbos import DBOS

app = Flask(__name__)
DBOS(flask=app)

@app.route("/")
@DBOS.workflow()
def test_workflow():
    return "<p>Workflow successful!</p>"

# After all decorators run, launch DBOS
DBOS.launch()

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

Assuming your file is main.py, run with python3 -m main (dev) or gunicorn -w 1 'main:app' -b 0.0.0.0:8000 (prod)

destroy

DBOS.destroy(
    destroy_registry: bool = False
)

Destroy the DBOS singleton, terminating all active workflows and closing database connections. After this completes, the singleton can be re-initialized. Useful for testing.

Parameters:

• destroy_registry: Whether to destroy the global registry of decorated functions. If set to True, destroy will "un-register" all decorated functions. You probably want to leave this False.

reset_system_database

DBOS.reset_system_database()

Destroy the DBOS system database, resetting DBOS's internal state in Postgres. Useful when testing a DBOS application to reset the internal state of DBOS between tests. For example, see its use in the testing tutorial. This is a destructive operation and should only be used in a test environment.

Configuration Management

load_config

load_config(
    config_file_path: str = "dbos-config.yaml",
    use_db_wizard: bool = True
) -> ConfigFile:

Load and parse a DBOS configuration file into a ConfigFile object.

Parameters:

• config_file_path: The path to the DBOS configuration file to parse.
• use_db_wizard: If the configuration file specifies default database connection parameters and there is no Postgres database there, whether to prompt the user to connect to a different database.

get_dbos_database_url

get_dbos_database_url(
    config_file_path: str = "dbos-config.yaml"
) -> str

Parse database connection information from a DBOS configuration file into a Postgres database connection string.

Parameters:

• config_file_path: The path to the DBOS configuration file to parse.