title: Dynamically create list of tasks
body: <p>I have a DAG which is created by querying DynamoDB for a list and for each item in the list a task is created using a PythonOperator and adding it to the DAG. Not show in the example below but it's important to note that some of the items on the list depend upon other tasks so I'm using <code>set_upstream</code> to enforce the dependencies.</p>
  
  <pre><code>- airflow_home
    \- dags
      \- workflow.py
  </code></pre>
  
  <p><strong>workflow.py</strong> </p>
  
  <pre><code>def get_task_list():
      # ... query dynamodb ...
  
  def run_task(task):
      # ... do stuff ...
  
  dag = DAG(dag_id='my_dag', ...)
  tasks = get_task_list()
  for task in tasks:
      t = PythonOperator(
          task_id=task['id'],
          provide_context=False,
          dag=dag,
          python_callable=run_task,
          op_args=[task]
      )
  </code></pre>
  
  <p>The problem is <code>workflow.py</code> is getting run over and over (every time a task runs?) and my <code>get_task_list()</code> method is getting throttled by AWS and throwing exceptions. </p>
  
  <p>I thought it was because whenever <code>run_task()</code> was called it was running all the globals in <code>workflow.py</code> so I've tried moving <code>run_task()</code> into a separate module like this:</p>
  
  <pre><code>- airflow_home
    \- dags
      \- workflow.py
      \- mypackage
        \- __init__
        \- task.py
  </code></pre>
  
  <p>But it didn't change anything. I've even tried putting <code>get_task_list()</code> into a SubDagOperator wrapped with a factory function, which still behaves the same way.</p>
  
  <p>Is my problem related to these issues?</p>
  
  <ul>
  <li><a href="https://stackoverflow.com/questions/44626883/tasks-added-to-dag-during-runtime-fail-to-be-scheduled/44636828#44636828">Tasks added to DAG during runtime fail to be scheduled</a></li>
  <li><a href="https://stackoverflow.com/questions/44636774/how-to-nest-an-airflow-dag-dynamically">How to nest an Airflow DAG dynamically?</a></li>
  </ul>
  
  <p>Also, why is <code>workflow.py</code> getting run so often and why would an error thrown by <code>get_task_list()</code> cause the individual task to fail when the task method doesn't reference <code>workflow.py</code> and has no dependencies on it?</p>
  
  <p>Most importantly, what would be the best way to both process the list in parallel and enforce any dependencies between items in the list?</p>