title: on_failure_callback in dynamic dags
body: <p>I'm creating dags dynamically from a list and would like to add an on_failure_callback to one of the tasks. I've tried the following code, but the callback doesn't seem to be getting executed. </p>
  
  <pre><code>dag_ids = ['dag_a', 'dag_b', 'dag_c']
  
  for dag_id in dag_ids:
  
      def failure_callback():
          logging.info('Inside failure callback for {}'.format(dag_id))
  
      def python_callable(dag_id):
          logging.info('Inside python callable for {}'.format(dag_id))
          raise Exception('Exception raised for dag_id {}'.format(dag_id))
  
      yesterday = datetime.datetime.combine(
          datetime.datetime.today() - datetime.timedelta(1),
          datetime.datetime.min.time())
  
      default_args = {
          'start_date': yesterday
      }    
  
      dag = models.DAG(
              dag_id,
              schedule_interval=None,
              catchup=False,
              default_args=default_args)
  
      with dag:
  
          python_task = PythonOperator(
              task_id='python_task',
              python_callable=python_callable,
              op_kwargs={'dag_id': dag_id},
              on_failure_callback=failure_callback,
              dag=dag)
  
          python_task
  
      globals()[dag_id] = dag
  </code></pre>
  
  <p>Any idea what I'm doing wrong here?</p>
  
  <p>Edit:</p>
  
  <p>Based on the suggestion, I passed the dag_id to the failure callback. But instead of dag_id, airflow passes the context dictionary. Any idea on how to pass an extra argument to failure callback besides the context dict?</p>
  
  <pre><code>ERROR - Inside failure callback for {u'next_execution_date': None, u'dag_run': &lt;DagRun dag_a @ 2019-02-19 19:23:54.006241: manual__2019-02-19T19:23:54.006241, externally triggered: True&gt;, u'tomorrow_ds_nodash': u'20190220', u'run_id': 'manual__2019-02-19T19:23:54.006241', u'test_mode': False, u'prev_execution_date': None, u'conf': &lt;module 'airflow.configuration' from '/usr/local/lib/airflow/airflow/configuration.py'&gt;, u'tables': None, u'task_instance_key_str': u'dag_a__python_task__20190219', u'END_DATE': '2019-02-19', u'execution_date': datetime.datetime(2019, 2, 19, 19, 23, 54, 6241), u'ts': '2019-02-19T19:23:54.006241', u'macros': &lt;module 'airflow.macros' from '/usr/local/lib/airflow/airflow/macros/__init__.py'&gt;, u'params': {}, u'ti': &lt;TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]&gt;, u'var': {u'json': None, u'value': None}, u'ds_nodash': u'20190219', u'dag': &lt;DAG: dag_a&gt;, u'end_date': '2019-02-19', u'latest_date': '2019-02-19', u'ds': '2019-02-19', u'task_instance': &lt;TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]&gt;, u'yesterday_ds_nodash': u'20190218', u'task': &lt;Task(PythonOperator): python_task&gt;, u'yesterday_ds': '2019-02-18', u'ts_nodash': u'20190219T192354.006241', u'tomorrow_ds': '2019-02-20'}
  </code></pre>
  
  <p>Referred to the question <a href="https://stackoverflow.com/questions/51851535/pass-other-arguments-to-on-failure-callback">here</a> and got it working!</p>