Awesome-omni-skill dbos-python

DBOS Python SDK for building reliable, fault-tolerant applications with durable workflows. Use this skill when writing Python code with DBOS, creating workflows and steps, using queues, using DBOSC...

install
source · Clone the upstream repo
git clone https://github.com/diegosouzapw/awesome-omni-skill
Claude Code · Install into ~/.claude/skills/
T=$(mktemp -d) && git clone --depth=1 https://github.com/diegosouzapw/awesome-omni-skill "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/development/dbos-python" ~/.claude/skills/diegosouzapw-awesome-omni-skill-dbos-python && rm -rf "$T"
manifest: skills/development/dbos-python/SKILL.md
source content

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

Rule Categories by Priority

PriorityCategoryImpactPrefix
1LifecycleCRITICAL
lifecycle-
2WorkflowCRITICAL
workflow-
3StepHIGH
step-
4QueueHIGH
queue-
5CommunicationMEDIUM
comm-
6PatternMEDIUM
pattern-
7TestingLOW-MEDIUM
test-
8ClientMEDIUM
client-
9AdvancedLOW
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