GeezerVison

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: DBOSConfig,
)

Parameters:

• config: Configuration parameters for DBOS. See the configuration docs.

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",
        "application_version": "0.1.0",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()
    dbos_workflow()

listen_queues​

DBOS.listen_queues(
    queues: Sequence[str]
)

Configure this DBOS process to only listen to (dequeue workflows from) specific queues. If this is not used, DBOS will listen to all registered queues. Must be called after DBOS is constructed and before it is launched, and may be called at most once.

Parameters:

• queues: The names of the queues to listen to.

destroy​

DBOS.destroy(
    *,
    destroy_registry: bool = False,
    workflow_completion_timeout_sec: int = 0,
)

Destroy the DBOS singleton, stopping its background threads (such as queue polling and the scheduler) and closing database connections. destroy waits up to workflow_completion_timeout_sec for active workflows to complete and cancels executor tasks that have not yet started.

Workflows still running after destroy will likely not be able to continue, because they rely on DBOS resources such as the system database and thread pool executor. Any errors they raise after destroy are safe, as they are not durably persisted. If your application has long-running workflows, we recommend shutting down your process shortly after calling destroy; its workflows will safely recover from their last completed step.

After destroy completes, DBOS can be re-launched: construct a new instance with DBOS(config=...) before calling DBOS.launch() again. This is 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.

migrate​

DBOS.migrate(
    system_database_url: str,
    *,
    schema: str = "dbos",
    application_role: Optional[str] = None,
    sharding: bool = False,
)

Create or migrate the DBOS system database without launching DBOS. This is the programmatic equivalent of the dbos migrate command. Run it with a privileged database role (for example, as part of your deployment's migration step), then configure your application with run_migrations=False so its role needs no DDL privileges. It can be called before or without constructing a DBOS instance.

Parameters:

• system_database_url: The system database to create or migrate. The database is created if it does not exist.
• schema: The Postgres schema containing the DBOS system tables. Defaults to dbos.
• application_role: A Postgres role to grant access to the DBOS system schema once it is migrated. Not supported for SQLite.
• sharding: Whether to create or migrate a sharded system database.

Example:

from dbos import DBOS

# In your migration script, run with a privileged role:
DBOS.migrate(os.environ["ADMIN_SYSTEM_DATABASE_URL"], application_role="my_app_role")

To migrate the tables used by datasources, use SQLAlchemyDatasource.migrate.

reset_system_database​

DBOS.reset_system_database(
    *,
    system_database_url: Optional[str] = None,
    schema: Optional[str] = None,
    truncate: bool = False,
)

Reset the DBOS system database, clearing DBOS's internal state. By default, this destroys the system database entirely; pass truncate=True to instead empty its tables, which is substantially faster. Useful when testing a DBOS application to reset the internal state of DBOS between tests. For example, see its use in the testing tutorial. It cannot be called after DBOS is launched. This is a destructive operation and should only be used in a test environment.

Parameters:

• system_database_url: The system database to reset. Defaults to the system database of the current DBOS instance. If no DBOS instance has been constructed, this must be supplied.
• schema: The Postgres schema containing the DBOS system tables. Defaults to the configured dbos_system_schema, or dbos. Only used when truncating.
• truncate: If True, empty the system tables instead of dropping the system database. This is recommended, as it is substantially faster. Because the migrated schema is left in place, DBOS does not need to re-create the system database on its next launch.