Conditional Workflows - Hatchet Documentation

Introduction

Hatchet V1 introduces the ability to add conditions to tasks in your workflows that determine whether or not a task should be run, based on a number of conditions. Conditions unlock a number of new ways to solve problems with Hatchet, such as:

A workflow that reads a feature flag, and then decides how to progress based on its value. In this case, you’d have two tasks that use parent conditions, where one task runs if the flag value is e.g. True, while the other runs if it’s False.
Any type of human-in-the-loop workflow, where you want to wait for a human to e.g. approve something before continuing the dag.

Types of Conditions

There are three types of Conditions in Hatchet V1:

Sleep conditions, which sleep for a specified duration before continuing
Event conditions, which wait for an event (and optionally a CEL expression evaluated on the payload of that event) before deciding how to continue
Parent conditions, which wait for a parent task to complete and then decide how to progress based on its output.

Or Groups

Conditions can also be combined using an Or operator into groups of conditions (called “or groups”) where at least one must be satisfied in order for the group to evaluate to True. An “or group” behaves like a boolean OR operator, where the group evaluates to True if at least one of its conditions is True.

Or groups are an extremely powerful feature because they let you express arbitrarily complex sets of conditions in conjunctive normal form (CNF) for determining when your tasks should run and when they should not. As a simple example, consider the following conditions:

Condition A: Checking if the output of a parent task is greater than 50
Condition B: Sleeping for 30 seconds
Condition C: Receiving the payment:processed event

You might want to progress in your workflow if A or B and C. In this case, we can express this set of conditions in CNF as A or B AND A or C where both A or B and A or C are or groups.

Usage

Conditions can be used at task declaration time in three ways:

They can be used in a wait_for fashion, where a task will wait for the conditions to evaluate to True before being run.
They can be used in a skip_if fashion, where a task will be skipped if the conditions evaluate to True.
They can be used in a cancel_if fashion, where a task will be cancelled if the conditions evaluate to True.

`wait_for`

Declaring a task with conditions to wait_for will cause the task to wait before starting for until its conditions evaluate to True. For instance, if you use wait_for with a 60 second sleep, the workflow will wait for 60 seconds before triggering the task. Similar, if the task is waiting for an event, it will wait until the event is fired before continuing.

`skip_if`

Declaring a task with conditions to skip_if will cause the task to be skipped if the conditions evaluate to True. For instance, if you use a parent condition to check if the output of a parent task is equal to some value, the task will be skipped if that condition evaluates to True.

`cancel_if`

Declaring a task with conditions to cancel_if will cause the task to be cancelled if the conditions evaluate to True. For instance, if you use a parent condition to check if the output of a parent task is equal to some value, the task will be cancelled if that condition evaluates to True.

⚠️

A task cancelled by a cancel_if operator will behave the same as any other cancellation in Hatchet, meaning that downstream tasks will be cancelled as well.

Example Workflow

In this example, we’re going to build the following workflow:

Branching DAG Workflow

Note the branching logic (left_branch and right_branch), as well as the use of skips and waits.

To get started, let’s declare the workflow.

examples/waits/worker.py


import random
from datetime import timedelta

from pydantic import BaseModel

from hatchet_sdk import (
    Context,
    EmptyModel,
    Hatchet,
    ParentCondition,
    SleepCondition,
    UserEventCondition,
    or_,
)

hatchet = Hatchet(debug=True)


class StepOutput(BaseModel):
    random_number: int


class RandomSum(BaseModel):
    sum: int


task_condition_workflow = hatchet.workflow(name="TaskConditionWorkflow")

src/v1/examples/dag_match_condition/complex-workflow.ts

import { Or, SleepCondition, UserEventCondition } from '@hatchet/v1/conditions';
import { ParentCondition } from '@hatchet/v1/conditions/parent-condition';
import { Context } from '@hatchet/step';
import { hatchet } from '../hatchet-client';

export const taskConditionWorkflow = hatchet.workflow({
  name: 'TaskConditionWorkflow',
});

examples/v1/workflows/complex-conditions.go

wf := factory.NewWorkflow[struct{}, TaskConditionWorkflowResult](
	create.WorkflowCreateOpts[struct{}]{
		Name: "TaskConditionWorkflow",
	},
	hatchet,
)

Next, we’ll start adding tasks to our workflow. First, we’ll add a basic task that outputs a random number:

examples/waits/worker.py

@task_condition_workflow.task()
def start(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

src/v1/examples/dag_match_condition/complex-workflow.ts

const start = taskConditionWorkflow.task({
  name: 'start',
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/v1/workflows/complex-conditions.go

start := wf.Task(
	taskOpts{
		Name: "start",
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

Next, we’ll add a task to the workflow that’s a child of the first task, but it has a wait_for condition that sleeps for 10 seconds.

examples/waits/worker.py

@task_condition_workflow.task(
    parents=[start], wait_for=[SleepCondition(timedelta(seconds=10))]
)
def wait_for_sleep(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

src/v1/examples/dag_match_condition/complex-workflow.ts

const waitForSleep = taskConditionWorkflow.task({
  name: 'waitForSleep',
  parents: [start],
  waitFor: [new SleepCondition('10s')],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/v1/workflows/complex-conditions.go

waitForSleep := wf.Task(
	taskOpts{
		Name:    "waitForSleep",
		Parents: []create.NamedTask{start},
		WaitFor: condition.SleepCondition(time.Second * 10),
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

This task will first wait for the parent task to complete, and then it’ll sleep for 10 seconds before executing and returning another random number.

Next, we’ll add a task that will be skipped on an event:

examples/waits/worker.py

@task_condition_workflow.task(
    parents=[start],
    wait_for=[SleepCondition(timedelta(seconds=30))],
    skip_if=[UserEventCondition(event_key="skip_on_event:skip")],
)
def skip_on_event(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

src/v1/examples/dag_match_condition/complex-workflow.ts

const skipOnEvent = taskConditionWorkflow.task({
  name: 'skipOnEvent',
  parents: [start],
  waitFor: [new SleepCondition('10s')],
  skipIf: [new UserEventCondition('skip_on_event:skip', 'true')],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/v1/workflows/complex-conditions.go

skipOnEvent := wf.Task(
	taskOpts{
		Name:    "skipOnEvent",
		Parents: []create.NamedTask{start},
		WaitFor: condition.SleepCondition(time.Second * 30),
		SkipIf:  condition.UserEventCondition("skip_on_event:skip", "true"),
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

In this case, our task will wait for a 30 second sleep, and then it will be skipped if the skip_on_event:skip is fired.

Next, let’s add some branching logic. Here we’ll add two more tasks, a left and right branch.

examples/waits/worker.py

@task_condition_workflow.task(
    parents=[wait_for_sleep],
    skip_if=[
        ParentCondition(
            parent=wait_for_sleep,
            expression="output.random_number > 50",
        )
    ],
)
def left_branch(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))


@task_condition_workflow.task(
    parents=[wait_for_sleep],
    skip_if=[
        ParentCondition(
            parent=wait_for_sleep,
            expression="output.random_number <= 50",
        )
    ],
)
def right_branch(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

src/v1/examples/dag_match_condition/complex-workflow.ts

const leftBranch = taskConditionWorkflow.task({
  name: 'leftBranch',
  parents: [waitForSleep],
  skipIf: [new ParentCondition(waitForSleep, 'output.randomNumber > 50')],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

const rightBranch = taskConditionWorkflow.task({
  name: 'rightBranch',
  parents: [waitForSleep],
  skipIf: [new ParentCondition(waitForSleep, 'output.randomNumber <= 50')],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/v1/workflows/complex-conditions.go

leftBranch := wf.Task(
	taskOpts{
		Name:    "leftBranch",
		Parents: []create.NamedTask{waitForSleep},
		SkipIf:  condition.ParentCondition(waitForSleep, "output.randomNumber > 50"),
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

rightBranch := wf.Task(
	taskOpts{
		Name:    "rightBranch",
		Parents: []create.NamedTask{waitForSleep},
		SkipIf:  condition.ParentCondition(waitForSleep, "output.randomNumber <= 50"),
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

These two tasks use the ParentCondition and skip_if together to check if the output of an upstream task was greater or less than 50, respectively. Only one of the two tasks will run: whichever one’s condition evaluates to True.

Next, we’ll add a task that waits for an event:

examples/waits/worker.py

@task_condition_workflow.task(
    parents=[start],
    wait_for=[
        or_(
            SleepCondition(duration=timedelta(minutes=1)),
            UserEventCondition(event_key="wait_for_event:start"),
        )
    ],
)
def wait_for_event(input: EmptyModel, ctx: Context) -> StepOutput:
    return StepOutput(random_number=random.randint(1, 100))

src/v1/examples/dag_match_condition/complex-workflow.ts

const waitForEvent = taskConditionWorkflow.task({
  name: 'waitForEvent',
  parents: [start],
  waitFor: [Or(new SleepCondition('1m'), new UserEventCondition('wait_for_event:start', 'true'))],
  fn: () => {
    return {
      randomNumber: Math.floor(Math.random() * 100) + 1,
    };
  },
});

examples/v1/workflows/complex-conditions.go

waitForEvent := wf.Task(
	taskOpts{
		Name:    "waitForEvent",
		Parents: []create.NamedTask{start},
		WaitFor: condition.Or(
			condition.SleepCondition(time.Minute),
			condition.UserEventCondition("wait_for_event:start", "true"),
		),
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		return &StepOutput{
			RandomNumber: rand.Intn(100) + 1,
		}, nil
	},
)

And finally, we’ll add the last task, which collects all of its parents and sums them up.

examples/waits/worker.py

@task_condition_workflow.task(
    parents=[
        start,
        wait_for_sleep,
        wait_for_event,
        skip_on_event,
        left_branch,
        right_branch,
    ],
)
def sum(input: EmptyModel, ctx: Context) -> RandomSum:
    one = ctx.task_output(start).random_number
    two = ctx.task_output(wait_for_event).random_number
    three = ctx.task_output(wait_for_sleep).random_number
    four = (
        ctx.task_output(skip_on_event).random_number
        if not ctx.was_skipped(skip_on_event)
        else 0
    )

    five = (
        ctx.task_output(left_branch).random_number
        if not ctx.was_skipped(left_branch)
        else 0
    )
    six = (
        ctx.task_output(right_branch).random_number
        if not ctx.was_skipped(right_branch)
        else 0
    )

    return RandomSum(sum=one + two + three + four + five + six)

Note that in this task, we rely on ctx.was_skipped to determine if a task was skipped.

src/v1/examples/dag_match_condition/complex-workflow.ts

taskConditionWorkflow.task({
  name: 'sum',
  parents: [start, waitForSleep, waitForEvent, skipOnEvent, leftBranch, rightBranch],
  fn: async (_, ctx: Context<any, any>) => {
    const one = (await ctx.parentOutput(start)).randomNumber;
    const two = (await ctx.parentOutput(waitForEvent)).randomNumber;
    const three = (await ctx.parentOutput(waitForSleep)).randomNumber;
    const four = (await ctx.parentOutput(skipOnEvent))?.randomNumber || 0;
    const five = (await ctx.parentOutput(leftBranch))?.randomNumber || 0;
    const six = (await ctx.parentOutput(rightBranch))?.randomNumber || 0;

    return {
      sum: one + two + three + four + five + six,
    };
  },
});

examples/v1/workflows/complex-conditions.go

wf.Task(
	taskOpts{
		Name: "sum",
		Parents: []create.NamedTask{
			start,
			waitForSleep,
			waitForEvent,
			skipOnEvent,
			leftBranch,
			rightBranch,
		},
	},
	func(ctx worker.HatchetContext, _ struct{}) (interface{}, error) {
		var startOutput StepOutput
		if err := ctx.ParentOutput(start, &startOutput); err != nil {
			return nil, err
		}

		var waitForSleepOutput StepOutput
		if err := ctx.ParentOutput(waitForSleep, &waitForSleepOutput); err != nil {
			return nil, err
		}

		var waitForEventOutput StepOutput
		ctx.ParentOutput(waitForEvent, &waitForEventOutput)

		// Handle potentially skipped tasks
		var skipOnEventOutput StepOutput
		var four int

		err := ctx.ParentOutput(skipOnEvent, &skipOnEventOutput)

		if err != nil {
			four = 0
		} else {
			four = skipOnEventOutput.RandomNumber
		}

		var leftBranchOutput StepOutput
		var five int

		err = ctx.ParentOutput(leftBranch, leftBranchOutput)
		if err != nil {
			five = 0
		} else {
			five = leftBranchOutput.RandomNumber
		}

		var rightBranchOutput StepOutput
		var six int

		err = ctx.ParentOutput(rightBranch, rightBranchOutput)
		if err != nil {
			six = 0
		} else {
			six = rightBranchOutput.RandomNumber
		}

		return &RandomSum{
			Sum: startOutput.RandomNumber + waitForEventOutput.RandomNumber +
				waitForSleepOutput.RandomNumber + four + five + six,
		}, nil
	},
)

This workflow demonstrates the power of the new conditional logic in Hatchet V1. You can now create complex workflows that are much more dynamic than workflows in the previous version of Hatchet, and do all of it declaratively (rather than, for example, by dynamically spawning child workflows based on conditions in the parent).

Directed Acyclic Graphs (DAGs)On Failure Tasks

We use cookies

Introduction

Types of Conditions

Or Groups

Usage

wait_for

skip_if

cancel_if

Example Workflow

`wait_for`

`skip_if`

`cancel_if`