Skip to main content

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(
    *,
    config: Optional[DBOSConfig] = None,
)

Parameters:

launch​

DBOS.launch()

Launch DBOS, initializing database connections and starting queues and scheduled workflows. Should be called after all decorators run. You should not run a DBOS workflow until after DBOS is launched.

Example:

import os
from dbos import DBOS, DBOSConfig

@DBOS.step()
def step_one():
    print("Step one completed!")

@DBOS.step()
def step_two():
    print("Step two completed!")

@DBOS.workflow()
def dbos_workflow():
    step_one()
    step_two()

# Configure and launch DBOS, then run a workflow.
if __name__ == "__main__":
    config: DBOSConfig = {
        "name": "dbos-starter",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()
    dbos_workflow()

listen_queues​

DBOS.listen_queues(
    queues: List[Queue]
)

Configure this DBOS process to only listen to (dequeue workflows from) specific queues. If this is not used, DBOS will listen to all declared queues. Must be called before DBOS is launched.

Parameters:

  • queues: The list of queues to listen to.

destroy​

DBOS.destroy(
    workflow_completion_timeout_sec: int = 0,
    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:

  • workflow_completion_timeout_sec: Wait this many seconds for active workflows to complete before shutting down.
  • 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.