title: Use DB to generate airflow tasks dynamically
body: <p>I want to run an airflow dag like so -></p>
  
  <ul>
  <li>I have 2 airflow workers W1 and W2. </li>
  <li>In W1 I have scheduled a single task (W1-1) but in W2, I want to create X number of tasks (W2-1, W2-2 ... W2-X). </li>
  <li>The number X and the bash command for each task  will be derived from a DB call.</li>
  <li>All tasks for worker W2 should run in parallel after W1 completes.</li>
  </ul>
  
  <p>This is my code</p>
  
  <pre><code>dag = DAG('deploy_single', catchup=False, default_args=default_args, schedule_interval='16 15 * * *')
  
  t1 = BashOperator(
          task_id='dummy_task',
          bash_command='echo hi &gt; /tmp/hi',
          queue='W1_queue',
          dag=dag)
  
  get_all_engines = "select full_command, queue_name from internal_airflow_hosts where logical_group = 'live_engines';"
  
  db_creds = json.loads(open('/opt/airflow/db_creds.json').read())
  conn_dict = db_creds["airflowdb_local"]
  connection = psycopg2.connect(**conn_dict)
  
  cursor = connection.cursor()
  
  cursor.execute(get_all_engines)
  records = cursor.fetchall()
  i = 1
  for record in records:
      t = BashOperator(
          task_id='script_test_'+str(i),
          bash_command="{full_command} ".format(full_command=str(record[0])),
          queue=str(record[1]),
          dag=dag)
      t.set_upstream(t1)
      i += 1
  
  cursor.close()
  connection.close()
  </code></pre>
  
  <p>However, when I run this, the task on W1 completed successfully but all tasks on W2 failed. In the airflow UI, I can see that it can resolve the correct number of tasks (10 in this case) but each of these 10 failed. </p>
  
  <p>Looking at the logs, I saw that on W2 (which is on a different machine), airflow could not find the <code>db_creds.json</code> file.</p>
  
  <p>I do not want to provide the DB creds file to W2. </p>
  
  <p>My question is how can an airflow task be created dynamically in this case?
  Basically i want to run a DB query on the airflow server and assign tasks to one or more workers based on the results of that query. The DB will contain updated info about which engines are active etc I want the DAG to reflect this. From logs, it looks like each worker runs the DB query. Providing access to DB to each worker is not an option.</p>