title: Can an Airflow task dynamically generate a DAG at runtime?
body: <p>I have an upload folder that gets irregular uploads. For each uploaded file, I want to spawn a DAG that is specific to that file.</p>
  <p>My first thought was to do this with a FileSensor that monitors the upload folder and, conditional on presence of new files, triggers a task that creates the separate DAGs. Conceptually:</p>
  <pre><code>Sensor_DAG (FileSensor -&gt; CreateDAGTask)
  
  |-&gt; File1_DAG (Task1 -&gt; Task2 -&gt; ...)
  |-&gt; File2_DAG (Task1 -&gt; Task2 -&gt; ...)
  </code></pre>
  <p>In my initial implementation, <code>CreateDAGTask</code> was a <code>PythonOperator</code> that created DAG globals, by placing them in the global namespace (<a href="https://stackoverflow.com/a/44127627/2114580">see this SO answer</a>), like so:</p>
  <pre class="lang-py prettyprint-override"><code>from airflow import DAG
  from airflow.operators.dummy_operator import DummyOperator
  from airflow.operators.python_operator import PythonOperator
  from airflow.contrib.sensors.file_sensor import FileSensor
  from datetime import datetime, timedelta
  from pathlib import Path
  
  UPLOAD_LOCATION = &quot;/opt/files/uploaded&quot;
  
  # Dynamic DAG generation task code, for the Sensor_DAG below
  def generate_dags_for_files(location=UPLOAD_LOCATION, **kwargs):
      dags = []
      for filepath in Path(location).glob('*'):
          dag_name = f&quot;process_{filepath.name}&quot;
          dag = DAG(dag_name, schedule_interval=&quot;@once&quot;, default_args={
              &quot;depends_on_past&quot;: True,
              &quot;start_date&quot;: datetime(2020, 7, 15),
              &quot;retries&quot;: 1,
              &quot;retry_delay&quot;: timedelta(hours=12)
          }, catchup=False)
          dag_task = DummyOperator(dag=dag, task_id=f&quot;start_{dag_name}&quot;)
  
          dags.append(dag)
  
          # Try to place the DAG into globals(), which doesn't work
          globals()[dag_name] = dag
  
      return dags
  </code></pre>
  <p>The main DAG then invokes this logic via a <code>PythonOperator</code>:</p>
  <pre><code># File-sensing DAG
  default_args = {
      &quot;depends_on_past&quot; : False,
      &quot;start_date&quot;      : datetime(2020, 7, 16),
      &quot;retries&quot;         : 1,
      &quot;retry_delay&quot;     : timedelta(hours=5),
  }
  with DAG(&quot;Sensor_DAG&quot;, default_args=default_args,
           schedule_interval= &quot;50 * * * *&quot;, catchup=False, ) as sensor_dag:
  
      start_task  = DummyOperator(task_id=&quot;start&quot;)
      stop_task   = DummyOperator(task_id=&quot;stop&quot;)
      sensor_task = FileSensor(task_id=&quot;my_file_sensor_task&quot;,
                               poke_interval=60,
                               filepath=UPLOAD_LOCATION)
      process_creator_task = PythonOperator(
          task_id=&quot;process_creator&quot;,
          python_callable=generate_dags_for_files,
      )
      start_task &gt;&gt; sensor_task &gt;&gt; process_creator_task &gt;&gt; stop_task
  </code></pre>
  <p>But that doesn't work, because by the time <code>process_creator_task</code> runs, the globals have already been parsed by Airflow. New globals after parse time are irrelevant.</p>
  <h2>Interim solution</h2>
  <p>Per <a href="https://stackoverflow.com/questions/39133376/airflow-dynamic-dag-and-task-ids">Airflow dynamic DAG and task Ids</a>, <s>I can achieve what I'm trying to do by omitting the <code>FileSensor</code> task altogether and just letting Airflow generate the per-file task at each scheduler heartbeat, replacing the Sensor_DAG with just executing <code>generate_dags_for_files</code>:</s> Update: Nevermind -- while this does create a DAG in the dashboard, actual execution runs into the <a href="https://stackoverflow.com/questions/57472942/dag-seems-to-be-missing">&quot;DAG seems to be missing&quot;</a> issue:</p>
  <pre class="lang-py prettyprint-override"><code>generate_dags_for_files()
  </code></pre>
  <p>This does mean that I can no longer regulate the frequency of folder polling with the <code>poke_interval</code> parameter of <code>FileSensor</code>; instead, Airflow will poll the folder every time it collects DAGs.</p>
  <p>Is that the best pattern here?</p>
  <h2>Other related StackOverflow threads</h2>
  <ul>
  <li><a href="https://stackoverflow.com/questions/54992541/run-airflow-dag-for-each-file">Run Airflow DAG for each file</a> and <a href="https://stackoverflow.com/questions/60082546/airflow-proper-way-to-run-dag-for-each-file">Airflow: Proper way to run DAG for each file</a>: identical use case, but the accepted answer uses two static DAGs, presumably with different parameters.</li>
  <li><a href="https://stackoverflow.com/questions/41517798/proper-way-to-create-dynamic-workflows-in-airflow">Proper way to create dynamic workflows in Airflow</a> - accepted answer dynamically creates tasks, not DAGs, via a complicated XCom setup.</li>
  </ul>