02 — Passing data with Context
A Context is the typed, persisted container that flows through a workflow. One task writes; a downstream task reads.
produce writes a value; consume depends on it and reads it.
#[workflow(name = "pipeline", description = "Pass data downstream")]
pub mod pipeline {
use super::*;
#[task]
pub async fn produce(ctx: &mut Context<serde_json::Value>) -> Result<(), TaskError> {
ctx.insert("numbers", serde_json::json!([1, 2, 3]))?;
Ok(())
}
#[task(dependencies = ["produce"])]
pub async fn consume(ctx: &mut Context<serde_json::Value>) -> Result<(), TaskError> {
let nums = ctx.get("numbers").cloned().unwrap_or_default();
ctx.insert("sum", serde_json::json!(/* sum of nums */ 6))?;
Ok(())
}
}
with cloaca.WorkflowBuilder("pipeline") as builder:
builder.description("Pass data downstream")
@cloaca.task()
def produce(context):
context.set("numbers", [1, 2, 3])
return context
@cloaca.task(dependencies=["produce"])
def consume(context):
nums = context.get("numbers")
context.set("sum", sum(nums))
return context
The dependencies list makes consume run after produce, so the value is
present when it reads. The context is persisted, so it survives a restart and is
returned as result.final_context.
- 03 — Dependencies & parallelism
- Reference: Context