@dbos-python

Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows. Use when adding DBOS to existing Python code, creating workflows and steps, or using queues for concurrency control.

DBOS Python Best Practices

Guide for building reliable, fault-tolerant Python applications with DBOS durable workflows.

When to Use

Reference these guidelines when:

  • 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

Runtime boundaries

Examples are version-sensitive fragments with omitted app setup/imports/handlers; verify installed DBOS APIs. Durable workflow identity does not make arbitrary external side effects exactly-once: steps may repeat after a crash before checkpointing. Use provider idempotency keys plus reconciliation for payments/messages. Authorize database targets and writes, protect serialized inputs/results, and never deserialize untrusted pickle data. API routes need authentication, tenant ownership, webhook signature/replay validation and bounded waits. Destructive test reset must use a dedicated disposable system database; a variable name alone is not target proof.

Rule Categories by Priority

PriorityCategoryImpactPrefix
1LifecycleCRITICALlifecycle-
2WorkflowCRITICALworkflow-
3StepHIGHstep-
4QueueHIGHqueue-
5CommunicationMEDIUMcomm-
6PatternMEDIUMpattern-
7TestingLOW-MEDIUMtest-
8ClientMEDIUMclient-
9AdvancedLOWadvanced-

Critical Rules

DBOS Configuration and Launch

A DBOS application MUST configure and launch DBOS inside its main function:

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

Workflows are comprised of steps. Any function performing complex operations or accessing external services must be a step:

@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

Read individual rule files for detailed explanations and examples:

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

References

Limitations

  • Use this skill only when the task clearly matches the scope described above.
  • Do not treat the output as a substitute for environment-specific validation, testing, or expert review.
  • Stop and ask for clarification if required inputs, permissions, safety boundaries, or success criteria are missing.