title: Airflow dynamic DAG and Task Ids
body: <p>I mostly see Airflow being used for ETL/Bid data related jobs. I'm trying to use it for business workflows wherein a user action triggers a set of dependent tasks in future. Some of these tasks may need to be cleared (deleted) based on certain other user actions.
  I thought the best way to handle this would be via dynamic task ids. I read that Airflow supports dynamic dag ids. So, I created a simple python script that takes DAG id and task id as command line parameters. However, I'm running into problems making it work. It  gives dag_id not found error. Has anyone tried this? Here's the code for the script (call it tmp.py) which I execute on command line as python (python tmp.py 820 2016-08-24T22:50:00 ):</p>
  
  <pre><code>from __future__ import print_function
  import os
  import sys
  import shutil
  from datetime import date, datetime, timedelta
  from airflow import DAG
  from airflow.operators.bash_operator import BashOperator
  execution = '2016-08-24T22:20:00'
  if len(sys.argv) &gt; 2  :
     dagid =  sys.argv[1]
     taskid = 'Activate' + sys.argv[1]
     execution = sys.argv[2]
  else:
     dagid = 'DAGObjectId'
     taskid = 'Activate'
  default_args = {'owner' : 'airflow', 'depends_on_past': False, 'start_date':date.today(), 'email': ['fake@fake.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1}
  dag = DAG(dag_id = dagid,
         default_args=default_args,
         schedule_interval='@once',
        )
   globals()[dagid] = dag
  task1 = BashOperator(
      task_id = taskid,
      bash_command='ls -l',
      dag=dag)
  
  fakeTask = BashOperator(
      task_id = 'fakeTask',
      bash_command='sleep 5',
      retries = 3,
      dag=dag)
  task1.set_upstream(fakeTask)
  
  airflowcmd = "airflow run " + dagid + " " + taskid + "  " + execution
  print("airflowcmd = " + airflowcmd)
  os.system(airflowcmd)
  </code></pre>