A Failure Stops What Depends on It in Python

Tasks fail at work: a source doesn't answer, a file holds a bad value. When clean_orders fails, build_report must not run: it would build the report from orders that were never cleaned, and send_email would mail that wrong report to the team. An orchestrator stops every task that waits for a failed task, directly or through other tasks, and still runs the rest.

The tasks a task waits for are its upstream. When Airflow skips a task because an upstream task didn't succeed, it gives it the status upstream_failed. The rule: before you run a task, look at the status of each task it waits for. If any of them is not "success" (it is "failed", or itself "upstream_failed"), mark the task "upstream_failed" and don't call its function. On another pipeline:

download   →  failed
unzip      →  upstream_failed   (waits for download)
load       →  upstream_failed   (waits for unzip)
make_logo  →  success           (waits for nothing)

Two Python tools help. status.get(name) reads a key like status[name] does, but gives None when the key isn't there yet, where status[name] raises a KeyError. continue ends this turn of a loop and jumps to the next one:

status = {"download": "failed"}
print(status.get("unzip"))    # None
for n in [1, 2, 3]:
    if n == 2:
        continue
    print(n)                  # prints 1, then 3

In this lesson's code, clean_orders raises ValueError("quantity is not a number: two"). Right now the runner prints the error but stores no status for it, and build_report and send_email run anyway. Your finished code must print:

extract_orders done
extract_customers done
clean_orders failed: quantity is not a number: two
extract_orders: success
extract_customers: success
clean_orders: failed
build_report: upstream_failed
send_email: upstream_failed

Then the check makes extract_customers fail instead, while clean_orders works. clean_orders must end success, and build_report and send_email must end upstream_failed again: build_report waits for both.

Your task: in run_pipeline, store "failed" for a task that raises. Then, at the top of the run-order loop, mark the task "upstream_failed" if any task it waits for didn't succeed, and skip it with continue.

Assignment
Follow the Coding Tutorial to make run_pipeline record failed tasks and skip every task whose upstream didn't succeed.

Hint
Compare with != "success", not == "failed": send_email waits for build_report, which is upstream_failed, not failed. And marking isn't enough: without continue, the try block still calls the task and writes "success" over the mark.