DBOS Class

from dbos import DBOS, DBOSConfig

config = DBOSConfig(...)
dbos = DBOS(config=config)

DBOS.launch()

DBOS.destroy(destroy_registry=False, workflow_completion_timeout_sec=0)

DBOS.reset_system_database()

handle = DBOS.start_workflow(func, *args, **kwargs)

@DBOS.workflow()
def process_order(order_id: str) -> str:
    return f"Processed {order_id}"

handle = DBOS.start_workflow(process_order, "order-123")
result = handle.get_result()  # Wait for completion

handle = await DBOS.start_workflow_async(func, *args, **kwargs)

handle = DBOS.retrieve_workflow(workflow_id, existing_workflow=True)

status = DBOS.get_workflow_status(workflow_id)

result = DBOS.get_result(workflow_id)

DBOS.cancel_workflow(workflow_id)

handle = DBOS.resume_workflow(workflow_id)

DBOS.delete_workflow(workflow_id, delete_children=False)

handle = DBOS.fork_workflow(workflow_id, start_step, application_version=None)

workflows = DBOS.list_workflows(
    workflow_ids=None,
    status=None,
    start_time=None,
    end_time=None,
    name=None,
    app_version=None,
    limit=None,
    offset=None,
    sort_desc=False
)

steps = DBOS.list_workflow_steps(workflow_id)

from typing import TypedDict, Optional, Any

class StepInfo(TypedDict):
    # The unique ID of the step in the workflow
    function_id: int
    # The (fully qualified) name of the step
    function_name: str
    # The step's output, if any
    output: Optional[Any]
    # The error the step threw, if any
    error: Optional[Exception]
    # If the step starts or retrieves the result of a workflow, its ID
    child_workflow_id: Optional[str]
    # The Unix epoch timestamp at which this step started
    started_at_epoch_ms: Optional[int]
    # The Unix epoch timestamp at which this step completed
    completed_at_epoch_ms: Optional[int]

from dbos import DBOS

@DBOS.step()
def step_one() -> str:
    return "result1"

@DBOS.step()
def step_two() -> str:
    return "result2"

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

handle = DBOS.start_workflow(my_workflow)
handle.get_result()

# List all steps
steps = DBOS.list_workflow_steps(handle.workflow_id)
for step in steps:
    print(f"Step {step['function_id']}: {step['function_name']}")
    print(f"  Output: {step['output']}")
    if step['error']:
        print(f"  Error: {step['error']}")
    if step['started_at_epoch_ms']:
        duration = step['completed_at_epoch_ms'] - step['started_at_epoch_ms']
        print(f"  Duration: {duration}ms")

completed_handle = DBOS.wait_first(handles, polling_interval_sec=1.0)

result = DBOS.run_step(dbos_step_options, func, *args, **kwargs)

@DBOS.workflow()
def my_workflow():
    # Execute a lambda as a step
    result = DBOS.run_step(None, lambda x: x * 2, 21)
    return result  # Returns 42

DBOS.sleep(seconds)

@DBOS.workflow()
def delayed_workflow():
    DBOS.sleep(60)  # Sleep for 60 seconds
    return "Done"

DBOS.send(destination_id, message, topic=None, idempotency_key=None)

message = DBOS.recv(topic=None, timeout_seconds=60)

DBOS.set_event(key, value)

value = DBOS.get_event(workflow_id, key, timeout_seconds=60)

events = DBOS.get_all_events(workflow_id)

DBOS.write_stream(key, value)

DBOS.close_stream(key)

for value in DBOS.read_stream(workflow_id, key):
    process(value)

wf_id = DBOS.workflow_id

step_id = DBOS.step_id

session = DBOS.sql_session

DBOS.logger.info("Processing order")

user = DBOS.authenticated_user

roles = DBOS.authenticated_roles

version = DBOS.application_version

exec_id = DBOS.executor_id

from typing import TypedDict

class VersionInfo(TypedDict):
    # Unique version identifier
    version_id: str
    # Human-readable version name
    version_name: str
    # Unix epoch timestamp when version was deployed
    version_timestamp: int
    # Unix epoch timestamp when version record was created
    created_at: int

from dbos import DBOS

# Get version info (internal API)
# This is typically used internally by DBOS
version_info = {
    "version_id": "v1.2.3-abc123",
    "version_name": "1.2.3",
    "version_timestamp": 1699999999000,
    "created_at": 1699999999000
}