Run code before all Airflow DAGs

Question

I'm using the relatively new airflow project. I have a bunch of DAGs written and running. Now I want to integrate a bug reporting service, so that if any code in any of the DAGs raises an exception, the information will be sent to a certain API. I can put the API call in the on_failure_callback of each DAG, but I need to execute an initializing line like bug_reporter.init(bug_reporter_token) that just needs to run once.

Is there a place in Airflow for initializing code? Right now I'm initializing the bug tracker at the beginning of every DAG definition file. This seems to be redundant, but I can't find a place to write a file that runs before the DAGs are defined. I've tried reading about plugins, but it doesn't seem to be there.

jhnclvr jhnclvr · Accepted Answer · 2017-03-29T20:03:51

In your DAG definition file, instead of a DAG use your own subclass:

from airflow.utils.decorators import apply_defaults
import bug_reporter

class DAGWithBugReporter(DAG):
    @apply_defaults
    def __init__(
        self,
        bug_reporter_token,
        *args, **kwargs):

        super(DAGWithBugReporter, self).__init__(*args, **kwargs)
        bug_reporter.init(bug_reporter_token)

Then in your dag definition:

dag = DAGWithBugReporter(
    dag_id='my_dag',
    schedule_interval=None,
    start_date=datetime(2017, 2, 26),
    bug_reporter_token=my_token_from_somewhere
)


t1 = PythonOperator(
    task_id='t1',
    provide_context=True,
    python_callable=my_callable,
    xcom_push=True,
    dag=dag)

Run code before all Airflow DAGs

1 Answers