What Airflow Adds in Python

Airflow is the orchestrator most data teams use. You describe a pipeline in a Python file called a DAG file. DAG stands for directed acyclic graph: tasks joined by arrows, with no loop, which is exactly what your deps dict holds. Here is the DAG file for the shop's pipeline; hidden code saves it as shop_dag.py before your code runs:

from datetime import datetime, timedelta

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator

import shop_tasks

with DAG(
    dag_id="shop_daily",
    schedule="@daily",
    start_date=datetime(2026, 5, 1),
    catchup=True,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
):
    extract_orders = PythonOperator(
        task_id="extract_orders",
        python_callable=shop_tasks.extract_orders,
    )
    extract_customers = PythonOperator(
        task_id="extract_customers",
        python_callable=shop_tasks.extract_customers,
    )
    clean_orders = PythonOperator(
        task_id="clean_orders",
        python_callable=shop_tasks.clean_orders,
    )
    build_report = PythonOperator(
        task_id="build_report",
        python_callable=shop_tasks.build_report,
    )
    send_email = PythonOperator(
        task_id="send_email",
        python_callable=shop_tasks.send_email,
    )

    extract_orders >> clean_orders
    clean_orders >> build_report
    extract_customers >> build_report
    build_report >> send_email

Each part is something you built by hand. schedule="@daily" makes one run per day. start_date with catchup=True backfills every missed day since May 1. "retries": 2 tries a failed task up to two more times. Each PythonOperator pairs a task_id with its function, like your tasks dict. And a >> b means b waits for a.

Reading it as text: Airflow can't run here, so you read the file line by line. "abc" in line is True when the text appears anywhere in the line, line.startswith(text) checks only the start, and line.split(sep) cuts the line at every sep into a list of pieces. Two names before = take one piece each.

line = 'owner="data-team",'
print(line.split('"'))               # ['owner=', 'data-team', ',']
left, right = "a -> b".split(" -> ")
print(right)                         # b

What Airflow adds on top of your loop: a scheduler that is always on and starts each run on time, at 2 a.m. with nobody there; a web page that shows every run and task with its status and log, with buttons to retry or backfill; and workers, machines that run the tasks, many at once. Its database keeps the record of runs. Your code does the same steps, but only while you run it yourself, and nobody else sees what failed.

The starting code has your deps, an empty dag_deps and a loop over the stripped lines of shop_dag.py; below the loop it prints both dicts side by side. When your loop works, it prints:

schedule="@daily",
default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
task: by hand | from the DAG file
extract_orders: [] | []
extract_customers: [] | []
clean_orders: ['extract_orders'] | ['extract_orders']
build_report: ['clean_orders', 'extract_customers'] | ['clean_orders', 'extract_customers']
send_email: ['build_report'] | ['build_report']
The same: True

Your task: inside the loop, print the lines that hold schedule= or retries, add the name from every task_id= line to dag_deps with an empty list, and for every >> line add the task on its left to the list of the task on its right.

Assignment
Follow the Coding Tutorial to read shop_dag.py with in, startswith and split, and rebuild its dependencies in dag_deps.

Hint
Every line is stripped first, so task_id="clean_orders", starts with task_id=. The >> lines come after all five tasks, so the task on the right is already a key of dag_deps when you add to its list.