Getting Started

Installation

Stabilize requires Python 3.11 or later.

pip install stabilize

For PostgreSQL support:

pip install stabilize[postgres]

Quick Start

Create a simple “Hello World” workflow.

from stabilize import (
    Workflow, StageExecution, TaskExecution,
    SqliteWorkflowStore, SqliteQueue, QueueProcessor, Orchestrator,
    Task, TaskResult, TaskRegistry,
)

# 1. Define a Task
class HelloTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        name = stage.context.get("name", "World")
        return TaskResult.success(outputs={"greeting": f"Hello, {name}!"})

# 2. Setup Infrastructure
store = SqliteWorkflowStore("sqlite:///:memory:", create_tables=True)
queue = SqliteQueue("sqlite:///:memory:", table_name="queue_messages")
queue._create_table()

# 3. Register Tasks and Create Processor
registry = TaskRegistry()
registry.register("hello", HelloTask)

processor = QueueProcessor(queue, store=store, task_registry=registry)

orchestrator = Orchestrator(queue)

# 4. Create & Run Workflow
workflow = Workflow.create(
    application="demo",
    name="Hello",
    stages=[
        StageExecution(
            ref_id="1",
            type="hello",
            name="Greet",
            tasks=[TaskExecution.create("Run", "hello", stage_start=True, stage_end=True)],
            context={"name": "Stabilize"},
        )
    ]
)

store.store(workflow)
orchestrator.start(workflow)
processor.process_all(timeout=5.0)

# 5. Check Result
result = store.retrieve(workflow.id)
print(result.stages[0].outputs["greeting"])
# Output: Hello, Stabilize!

Next Steps

• Enable event sourcing for audit trails and replay — see Event Sourcing.

• Explore built-in tasks: ShellTask, HTTPTask, PythonTask, DockerTask — see Tasks.

• Learn about parallel stages and DAG execution — see Core Concepts.