Workflow
The Workflow class represents a built workflow that can be executed by a runner. Workflows are typically created using the WorkflowBuilder class.
A workflow defines the structure and execution order of tasks, including their dependencies and metadata. Once built, workflows are immutable and can be executed multiple times with different contexts.
Workflows are created using the WorkflowBuilder:
import cloaca
# Define tasks
@cloaca.task()
def task_a(context):
context.set("step", "A completed")
return context
@cloaca.task(dependencies=["task_a"])
def task_b(context):
previous_step = context.get("step")
context.set("step", f"{previous_step}, B completed")
return context
# Create workflow
builder = cloaca.WorkflowBuilder("my_workflow")
builder.description("Example workflow with dependencies")
builder.add_task("task_a")
builder.add_task("task_b")
workflow = builder.build()
@cloaca.workflow_params(...) declares a workflow’s typed, injectable
execute-time inputs — the Python parity of Rust’s #[workflow(params(...))].
Apply it to the workflow’s entry task. Each entry is name=Type (required) or
name=(Type, default) (optional). The compiler turns these into JSON-Schema
InputSlots exposed on the workflow’s declared_params and rendered as a typed
form in the web UI’s Run dialog.
@cloaca.workflow_params(
source_id=str, # required
batch_size=(int, 500), # optional, default 500
)
@cloaca.task(dependencies=[])
def prepare(context):
return context
| Form | Meaning |
|---|---|
name=Type |
Required param (Type is str / int / float / bool). |
name=(Type, default) |
Optional param with a default. |
Declared params are validated at the execute API — supplying an unknown or
mistyped value, or omitting a required one, is rejected. They are otherwise a
pass-through into the run context. See
Declare workflow inputs for the full
flow (Rust + Python + the declared_params API surface).
Trigger-/cron-fired workflows are executed with no caller-supplied params, so declare those workflows’ params with defaults (optional) to keep them firing unattended.
The built Workflow object exposes three read-only properties and five
inspection methods (crates/cloacina-python/src/workflow.rs, PyWorkflow):
Properties:
name(str): Unique identifier for the workflowdescription(str): Human-readable description of the workflow’s purposeversion(str): Content-derived workflow version
Methods:
topological_sort()→list[str]— task names in a valid execution orderget_execution_levels()→list[list[str]]— tasks grouped by parallel-executable levelget_roots()→list[str]— tasks with no dependenciesget_leaves()→list[str]— tasks nothing depends onvalidate()— raisesValueErrorif the workflow structure is invalid
# Get workflow information
print(f"Workflow name: {workflow.name}")
print(f"Description: {workflow.description}")
print(f"Version: {workflow.version}")
print(f"Execution order: {workflow.topological_sort()}")
print(f"Parallel levels: {workflow.get_execution_levels()}")
Workflows are executed using a DefaultRunner:
# Create runner
runner = cloaca.DefaultRunner("sqlite:///:memory:")
# Register workflow
cloaca.register_workflow_constructor("my_workflow", lambda: workflow)
# Execute workflow
context = cloaca.Context({"input_data": "example"})
result = runner.execute("my_workflow", context)
print(f"Execution status: {result.status}")
print(f"Final context: {result.final_context.to_dict()}")
Workflows are automatically validated during the build process:
try:
workflow = builder.build()
print("Workflow is valid")
except ValueError as e:
print(f"Validation failed: {e}")
Common validation errors include:
- Circular dependencies between tasks
- Missing task dependencies
- Duplicate task IDs
- Invalid task references
- WorkflowBuilder - Build workflows
- Task Decorator - Define workflow tasks
- DefaultRunner - Execute workflows
- Context - Data flow between tasks
- Tutorials - For worked, end-to-end examples (ETL, parallel fan-in, error handling), see the embedded tutorials — in particular Dependencies and Error Handling.