We use cookies

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. There are three types of Conditions in Hatchet V1:

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

These conditions can also be combined using an Or operator into groups of conditions where at least one must be satisfied in order for the group to evaluate to True.

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

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

Use Cases

There are a number of use cases that these features unlock. Some examples might be:

1. 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.
2. Any type of human-in-the-loop workflow, where you want to wait for a human to e.g. approve something before continuing the run.

Example Workflow

In this example, we’re going to build the following 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.

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")

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

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

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.

@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))

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:

@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))

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.

@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))

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:

@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))

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

@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.

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).