Marketplace 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 DBOSClient from external applications, or building applications that need to be resilient to failures.

install
source · Clone the upstream repo
git clone https://github.com/aiskillstore/marketplace
Claude Code · Install into ~/.claude/skills/
T=$(mktemp -d) && git clone --depth=1 https://github.com/aiskillstore/marketplace "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/sickn33/dbos-python" ~/.claude/skills/aiskillstore-marketplace-dbos-python && rm -rf "$T"
manifest: skills/sickn33/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