界面已本地化;技能介绍与说明文档按所选语言由 AI 自动翻译,代码、命令、路径和专业术语保留原文,以原作者文档为准。
探索技能
开发与工程 · SKILL.md

dbos-python

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.

独立 GitHub 快照 · 自动分类 · 以原作者文档为准

技能说明

GitHub

name: dbos-python description: "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." risk: safe source: "https://docs.dbos.dev/" date_added: "2026-02-27"

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
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.
我的收藏