Antigravity-awesome-skills 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/benjaminasterA/antigravity-awesome-skills
Claude Code · Install into ~/.claude/skills/
T=$(mktemp -d) && git clone --depth=1 https://github.com/benjaminasterA/antigravity-awesome-skills "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/dbos-python" ~/.claude/skills/benjaminastera-antigravity-awesome-skills-dbos-python && rm -rf "$T"
manifest:
skills/dbos-python/SKILL.mdsource 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
| Priority | Category | Impact | Prefix |
|---|---|---|---|
| 1 | Lifecycle | CRITICAL | |
| 2 | Workflow | CRITICAL | |
| 3 | Step | HIGH | |
| 4 | Queue | HIGH | |
| 5 | Communication | MEDIUM | |
| 6 | Pattern | MEDIUM | |
| 7 | Testing | LOW-MEDIUM | |
| 8 | Client | MEDIUM | |
| 9 | Advanced | LOW | |
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
orDBOS.start_workflow
from a stepDBOS.recv - Do NOT use threads to start workflows - use
or queuesDBOS.start_workflow - 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