on_failure_callback in dynamic dags

Question

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.

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

Any idea what I'm doing wrong here?

Edit:

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?

ERROR - Inside failure callback for {u'next_execution_date': None, u'dag_run': <DagRun dag_a @ 2019-02-19 19:23:54.006241: manual__2019-02-19T19:23:54.006241, externally triggered: True>, 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': <module 'airflow.configuration' from '/usr/local/lib/airflow/airflow/configuration.py'>, 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': <module 'airflow.macros' from '/usr/local/lib/airflow/airflow/macros/__init__.py'>, u'params': {}, u'ti': <TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]>, u'var': {u'json': None, u'value': None}, u'ds_nodash': u'20190219', u'dag': <DAG: dag_a>, u'end_date': '2019-02-19', u'latest_date': '2019-02-19', u'ds': '2019-02-19', u'task_instance': <TaskInstance: dag_a.python_task 2019-02-19 19:23:54.006241 [failed]>, u'yesterday_ds_nodash': u'20190218', u'task': <Task(PythonOperator): python_task>, u'yesterday_ds': '2019-02-18', u'ts_nodash': u'20190219T192354.006241', u'tomorrow_ds': '2019-02-20'}

Referred to the question here and got it working!

See if you can draw some hints from [this](https://stackoverflow.com/a/54599722/3679900) — y2k-shubham, Feb 18 '19 at 06:34

score 2 · Accepted Answer · answered Feb 19 '19 at 20:09

Used partials package to get it working. The updated code is given below.

from functools import partial

dag_ids = ['dag_a', 'dag_b', 'dag_c']

for dag_id in dag_ids:

    def failure_callback(dag_id, context):
        logging.error('Inside failure callback for {}'.format(dag_id))

    def python_callable(dag_id):
        logging.error('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,
            # Continue to run DAG once per day
            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=partial(failure_callback, dag_id),
            dag=dag)

        python_task

    globals()[dag_id] = dag

score 0 · Answer 2 · answered Feb 18 '19 at 22:39

0

Inside the for loop, just pass the dag_id to the failure_callback and then you should be able to see the logs from the failure_callback().

def failure_callback(dag_id):
    logging.info('Inside failure callback for {}'.format(dag_id))

answered Feb 18 '19 at 22:39

Ryan Yuan

2,396
2
13
23

I've updated the question after I tried passing dag_id to the callback. – blessonm Feb 19 '19 at 20:00

on_failure_callback in dynamic dags

2 Answers2