DBOS Python Best Practices

When to Apply

• Adding DBOS to existing Python code
• Creating workflows and steps
• Using queues for concurrency control
• Implementing workflow communication (events, messages, streams)
• Configuring and launching DBOS applications
• Using DBOSClient from external applications
• Testing DBOS applications

Rule Categories by Priority

lifecycle-

workflow-

step-

queue-

comm-

pattern-

test-

client-

advanced-

Critical Rules

DBOS Configuration and Launch

import os
from dbos import DBOS, DBOSConfig

@DBOS.workflow()
def my_workflow():
    pass

if __name__ == "__main__":
    config: DBOSConfig = {
        "name": "my-app",
        "system_database_url": os.environ.get("DBOS_SYSTEM_DATABASE_URL"),
    }
    DBOS(config=config)
    DBOS.launch()

Workflow and Step Structure

@DBOS.step()
def call_external_api():
    return requests.get("https://api.example.com").json()

@DBOS.workflow()
def my_workflow():
    result = call_external_api()
    return result

Key Constraints

• Do NOT call

  DBOS.start_workflow

  or

  DBOS.recv

  from a step
• Do NOT use threads to start workflows - use

  DBOS.start_workflow

  or queues
• Workflows MUST be deterministic - non-deterministic operations go in steps
• Do NOT create/update global variables from workflows or steps

How to Use

references/lifecycle-config.md
references/workflow-determinism.md
references/queue-concurrency.md

References

• https://docs.dbos.dev/
• https://github.com/dbos-inc/dbos-transact-py