Dynamic operators within DAGs (Airflow) I would like to know if what I did to achieve to goal of dynamic operators within an Airflow DAG (Directed Acyclic Graph) is a good or a bad practice. 

The goal I had to achieve was:


  Create a 'x' amount of operators within
  a DAG based on the result of an API call. This DAG will run for example every week.


The program that I made works, but I don't know if it is a good practice for developing DAGs for Airflow. So my question is: is it a good practice or if not what would be a better solution to this problem? 

The code I used to achieve this goal:

with DAG('my_dag', default_args=default_args, schedule_interval='@weekly') as my_dag:

    start = DummyOperator(
        task_id='start'
    )

    endpoint = 'www.example.com/api/dummies'
    r = requests.get(endpoint)
    dummies = r.json()

   for _, dummy in enumerate(dummies):
        dummy_operator = DummyOperator(
                    task_id='dummy_opr_{}'.format(dummy['id']
                    )
        start &gt;&gt; dummy_operator
It is a good practice for Airflow. Moreover, Airflow positioning it as one of their "killer-features". It is good to create dynamic tasks in DAG, it is pretty OK to create dynamically created DAGs. One thing I recommend you to think about is that it is not good to create hundreds and thousands of one-time-running tasks because it will lead to:


Filling your Airflow database with huge amount of garbage data
Inability to analyze DAGs visually. Here is the example of Airflow tree tab of tasks like these:




Here is the sample code for dynamic DAG creation:

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator


def create_dag(dag_id, schedule, dag_number, default_args):

    def hello_world_py(*args):
        print('Hello World')
        print('This is DAG: {}'.format(str(dag_number)))

    dag = DAG(dag_id,
              schedule_interval=schedule,
              default_args=default_args)

    with dag:
        t1 = PythonOperator(
            task_id='hello_world',
            python_callable=hello_world_py,
            dag_number=dag_number)

    return dag


# build a dag for each number in range(10)
for n in range(1, 10):
    dag_id = 'hello_world_{}'.format(str(n))

    default_args = {'owner': 'airflow',
                    'start_date': datetime(2018, 1, 1)
                    }

    schedule = '@daily'

    dag_number = n

    globals()[dag_id] = create_dag(dag_id,
                                  schedule,
                                  dag_number,
                                  default_args)


You can read more about it here.