superdag.py

from airflow.operators.python import task
from airflow.utils.dates import days_ago
from airflow.decorators import dag
from airflow.operators.subdag import SubDagOperator

# uninteresting args
default_args = {
    "schedule_interval": None,
    "start_date": days_ago(1),
    "default_args": {"owner": "airflow"},
    "catchup": False,
}

# this is our subdag
@dag(**default_args)
def subdag():
    @task
    def say_hi():
        print("hi")

    inner_task = say_hi()


# this is the superdag
@dag(**default_args)
def superdag():
    outer_task = SubDagOperator(
        task_id="outertask",
        dag_id="superdag.outertask",  # <-- line 29
        subdag=subdag(),
    )

super_dag = superdag()

# if line 29 is commented out:
#
# File "/Users/matt/src/airflow/airflow/operators/subdag.py", line 88, in _validate_dag
#    raise AirflowException(
# airflow.exceptions.AirflowException: The subdag's dag_id should have the form:
# '{parent_dag_id}.{this_task_id}' expected 'superdag.outertask'; received 'subdag'.

# if line 29 is included:
#
# File "/Users/matt/src/airflow/airflow/models/baseoperator.py", line 508, in __init__
#    raise AirflowException(
# airflow.exceptions.AirflowException: Invalid arguments were passed to SubDagOperator (task_id: outertask).
#  Invalid arguments were: **kwargs: {'dag_id': 'superdag.outertask'}