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.

31,392 stars
Complexity: easy

About this skill

This skill equips an AI agent with comprehensive best practices and guidelines for developing robust, fault-tolerant Python applications leveraging DBOS durable workflows. It covers key aspects from integrating DBOS into existing Python projects, designing workflows and individual steps, and managing concurrency using queues. The agent can use this knowledge to assist developers in building resilient systems, generating correct DBOS-compliant code, and understanding advanced features like workflow communication (events, messages, streams), application configuration, and external client interactions. This empowers the agent to serve as a knowledgeable assistant for Python developers working with the DBOS framework.

Best use case

Assisting developers in designing and implementing reliable, fault-tolerant Python applications. Generating code snippets and architectural advice adhering to DBOS best practices. Guiding through DBOS integration, workflow creation, concurrency control, and testing.

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.

The AI agent provides accurate, contextually relevant advice, code examples, or architectural recommendations for building Python applications with DBOS, resulting in more reliable and fault-tolerant software designs and implementations.

Practical example

Example input

How do I create a durable workflow in Python using DBOS for a multi-step order processing system, ensuring fault tolerance and idempotency?

Example output

To create a durable workflow for an order processing system with DBOS, you'd define a class decorated with `@dbos_workflow()`. Each processing step (e.g., payment, inventory update, shipping) should be a `@transactional_function()` within the workflow. This ensures that each step is executed exactly once, even if failures occur. Here's a basic structure:

```python
from dbos_core import Workflow, WorkflowContext, transactional_function, dbos_workflow

@dbos_workflow()
class OrderProcessingWorkflow(Workflow):
    def __init__(self, ctx: WorkflowContext):
        super().__init__(ctx)

    @transactional_function()
    def _process_payment(self, order_id: str, amount: float) -> bool:
        """Simulates processing a payment. Idempotent by design."""
        self.ctx.logger.info(f"Processing payment for order {order_id}, amount {amount}")
        # Add actual payment gateway interaction logic here
        # Ensure this function is idempotent (e.g., check if payment already processed)
        return True # Or False on failure

    @transactional_function()
    def _update_inventory(self, order_id: str, items: dict) -> bool:
        """Simulates updating inventory. Idempotent."""
        self.ctx.logger.info(f"Updating inventory for order {order_id}, items {items}")
        # Add actual inventory system interaction logic here
        # Ensure idempotency
        return True

    @transactional_function()
    def _send_confirmation_email(self, order_id: str, email: str) -> bool:
        """Sends a confirmation email. Can be retried."""
        self.ctx.logger.info(f"Sending confirmation email for order {order_id} to {email}")
        # Add email sending logic here
        return True

    def run(self, order_details: dict) -> str:
        order_id = order_details['id']
        self.ctx.logger.info(f"Starting order workflow for {order_id}")

        if not self._process_payment(order_id, order_details['total_amount']):
            self.ctx.logger.error(f"Payment failed for {order_id}")
            return "ORDER_PAYMENT_FAILED"

        if not self._update_inventory(order_id, order_details['items']):
            self.ctx.logger.error(f"Inventory update failed for {order_id}")
            # Compensation or manual intervention might be needed here
            return "ORDER_INVENTORY_FAILED"

        self._send_confirmation_email(order_id, order_details['customer_email'])
        self.ctx.logger.info(f"Order {order_id} processed successfully.")
        return "ORDER_PROCESSED_SUCCESSFULLY"

# To invoke this workflow from your DBOS application:
# workflow_instance = OrderProcessingWorkflow(dbos_runtime_context) # Context provided by DBOS
# result = await workflow_instance.run({
#     'id': 'ORD-2023-001',
#     'total_amount': 99.99,
#     'items': {'itemA': 1, 'itemB': 2},
#     'customer_email': 'customer@example.com'
# })
# print(result)
```

**Key DBOS Principles Applied:**
*   **Durable Workflow:** The `run` method orchestrates the steps, and DBOS ensures its state is preserved across failures.
*   **Transactional Functions:** Each `@transactional_function()` (`_process_payment`, `_update_inventory`, `_send_confirmation_email`) is executed within a transaction, providing atomicity and exactly-once semantics. If a step fails, DBOS can retry it or roll back, maintaining data consistency.
*   **Idempotency:** It's crucial that your business logic within `@transactional_function()` is idempotent, meaning executing it multiple times has the same effect as executing it once (e.g., only debiting a payment once even if the function is retried).

When to use this skill

  • When an AI agent needs to provide guidance or generate code for: - Integrating DBOS into new or existing Python applications. - Designing and implementing durable workflows and individual steps. - Managing concurrent operations using DBOS queues. - Setting up inter-workflow communication via events, messages, or streams. - Configuring and launching DBOS-powered Python applications. - Developing external applications that interact with DBOS using DBOSClient. - Crafting test strategies for DBOS applications.

When not to use this skill

  • When the task does not involve Python development, durable workflows, or the DBOS framework specifically. This skill is not for direct execution of Python code by the agent, but rather a knowledge base for guiding and generating code for Python development with DBOS.

Installation

Claude Code / Cursor / Codex

$curl -o ~/.claude/skills/dbos-python/SKILL.md --create-dirs "https://raw.githubusercontent.com/sickn33/antigravity-awesome-skills/main/plugins/antigravity-awesome-skills-claude/skills/dbos-python/SKILL.md"

Manual Installation

  1. Download SKILL.md from GitHub
  2. Place it in .claude/skills/dbos-python/SKILL.md inside your project
  3. Restart your AI agent — it will auto-discover the skill

How dbos-python Compares

Feature / Agentdbos-pythonStandard Approach
Platform SupportClaudeLimited / Varies
Context Awareness High Baseline
Installation ComplexityeasyN/A

Frequently Asked Questions

What does this skill do?

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.

Which AI agents support this skill?

This skill is designed for Claude.

How difficult is it to install?

The installation complexity is rated as easy. You can find the installation instructions above.

Where can I find the source code?

You can find the source code on GitHub using the link provided at the top of the page.

Related Guides

SKILL.md Source

# 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 | `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:

```python
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:

```python
@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

Related Skills

n8n-expression-syntax

31392
from sickn33/antigravity-awesome-skills

Validate n8n expression syntax and fix common errors. Use when writing n8n expressions, using {{}} syntax, accessing $json/$node variables, troubleshooting expression errors, or working with webhook data in workflows.

Developer ToolsClaude

mermaid-expert

31392
from sickn33/antigravity-awesome-skills

Create Mermaid diagrams for flowcharts, sequences, ERDs, and architectures. Masters syntax for all diagram types and styling.

Developer ToolsClaude

mcp-builder-ms

31392
from sickn33/antigravity-awesome-skills

Use this skill when building MCP servers to integrate external APIs or services, whether in Python (FastMCP) or Node/TypeScript (MCP SDK).

Developer ToolsClaude

makepad-deployment

31392
from sickn33/antigravity-awesome-skills

CRITICAL: Use for Makepad packaging and deployment. Triggers on: deploy, package, APK, IPA, 打包, 部署, cargo-packager, cargo-makepad, WASM, Android, iOS, distribution, installer, .deb, .dmg, .nsis, GitHub Actions, CI, action, marketplace

Developer ToolsClaude

macos-menubar-tuist-app

31392
from sickn33/antigravity-awesome-skills

Build, refactor, or review SwiftUI macOS menubar apps that use Tuist.

Developer ToolsClaude

kaizen

31392
from sickn33/antigravity-awesome-skills

Guide for continuous improvement, error proofing, and standardization. Use this skill when the user wants to improve code quality, refactor, or discuss process improvements.

Developer ToolsClaude

issues

31392
from sickn33/antigravity-awesome-skills

Interact with GitHub issues - create, list, and view issues.

Developer ToolsClaude

hugging-face-tool-builder

31392
from sickn33/antigravity-awesome-skills

Your purpose is now is to create reusable command line scripts and utilities for using the Hugging Face API, allowing chaining, piping and intermediate processing where helpful. You can access the API directly, as well as use the hf command line tool.

Developer ToolsClaude

git-pushing

31392
from sickn33/antigravity-awesome-skills

Stage all changes, create a conventional commit, and push to the remote branch. Use when explicitly asks to push changes ("push this", "commit and push"), mentions saving work to remote ("save to github", "push to remote"), or completes a feature and wants to share it.

Developer ToolsClaude

git-hooks-automation

31392
from sickn33/antigravity-awesome-skills

Master Git hooks setup with Husky, lint-staged, pre-commit framework, and commitlint. Automate code quality gates, formatting, linting, and commit message enforcement before code reaches CI.

Developer ToolsClaude

gh-review-requests

31392
from sickn33/antigravity-awesome-skills

Fetch unread GitHub notifications for open PRs where review is requested from a specified team or opened by a team member. Use when asked to "find PRs I need to review", "show my review requests", "what needs my review", "fetch GitHub review requests", or "check team review queue".

Developer ToolsClaude

fp-types-ref

31392
from sickn33/antigravity-awesome-skills

Quick reference for fp-ts types. Use when user asks which type to use, needs Option/Either/Task decision help, or wants fp-ts imports.

Developer ToolsClaude