dbos-python

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

Priority	Category	Impact	Prefix
1	Lifecycle	CRITICAL	lifecycle-
2	Workflow	CRITICAL	workflow-
3	Step	HIGH	step-
4	Queue	HIGH	queue-
5	Communication	MEDIUM	comm-
6	Pattern	MEDIUM	pattern-
7	Testing	LOW-MEDIUM	test-
8	Client	MEDIUM	client-
9	Advanced	LOW	advanced-

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

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

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.