Human Approval Steps

Concepts covered: Sensors and deferral

Some steps belong to a person. A pricing model that moves 8 percent of listings by more than 5 percent goes live only when an analyst says so, and the analyst wants to see the numbers first. Before Airflow 3.1 the usual workaround was a task that failed on purpose until someone cleared it, with the decision living in a chat thread rather than in the run. The standard provider now ships operators that put the question inside the run: the task posts a subject and a body, waits for one of a fixed set of answers, records who gave it and when, and carries the answer into the tasks after it.

from datetime import timedelta

from airflow.providers.standard.operators.hitl import ApprovalOperator

approve_prices = ApprovalOperator(
    task_id="approve_prices",
    subject="Publish tonight's prices?",
    body="Model v12 moves 8 percent of listings by more than 5 percent.",
    defaults="Reject",
    execution_timeout=timedelta(hours=4),
)

score_listings >> approve_prices >> publish_prices

When approve_prices starts, it records the question and leaves the worker: the cell turns to deferred, a state that means the task is waiting on an event outside any worker, with its wait held by the triggerer, a long-lived Airflow process beside the scheduler. A later lesson covers that state in depth; for now, read it as a wait that costs no worker slot, so 30 runs waiting on 30 analysts occupy nothing. The question appears on the Required Actions page of the UI, showing the subject and body above its 2 buttons, next to the run it belongs to. Any user with access to the DAG can answer, unless assigned_users names who may. The task instance stores the answer along with who gave it and when.

Reject ends the task in success, because the task did its job of asking, and skips the tasks after it. By default ignore_downstream_trigger_rules is False: it skips the tasks directly downstream, here publish_prices. Tasks further down follow their own trigger rules, so a notify_team after publish_prices with trigger_rule="all_done" still runs on a Reject. Setting it to True skips every task downstream whatever its trigger rule, as a short-circuit does by default. fail_on_reject=True makes Reject fail the task instead, for a run that should turn red when the answer is no. Approve returns and the chain continues. The option the person chose is stored with the task, so a task further down can read which answer was given, which belongs to the lessons on passing values between tasks.

When nobody answers

execution_timeout bounds the wait, the same argument that bounds any task. With defaults set, the timeout picks that answer and the task ends as if a person had chosen it, which is the shape for a decision with a safe side: an unreviewed price list stays unpublished. Without defaults, the timeout fails the task, and the retry policy from the first section applies to it like any failure, so retries=1 asks the question a second time 4 hours later. Clearing the task from the grid asks again at once, with a fresh 4 hours.

More than yes or no

HITLBranchOperator offers a list of task ids as its options and runs the one the person picks, the branching pattern from the intermediate lesson with a person in place of the function. HITLOperator takes any list of options, and multiple=True lets the person pick several. HITLEntryOperator collects typed values through form fields. subject and body work the same way in all 4, and so do defaults, execution_timeout, notifiers and assigned_users.

from datetime import timedelta

from airflow.providers.standard.operators.hitl import HITLBranchOperator

choose_rollout = HITLBranchOperator(
    task_id="choose_rollout",
    subject="How far should model v12 roll out?",
    options=["publish_all", "publish_sample", "hold"],
    defaults="hold",
    execution_timeout=timedelta(hours=4),
)

choose_rollout >> [publish_all, publish_sample, hold]

A question on a page nobody opens waits for its timeout. notifiers takes a list of notifier objects, the kind a failure callback can be, and runs them when the task starts waiting, passing them the subject and body along with a link to the response page. A chat notifier from a provider puts the question in the analysts' channel at 05:10 with a button that opens the page, and the 4 hour clock runs from then. The timeout should match how the team works: 4 hours covers a morning, and a question posted at 02:00 with a 4 hour timeout answers itself before anyone is awake.

About this lesson

This section is part of the Task Lifecycle: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.