add support for airflow emails

Author: OriHoch

airflow/dags/committees.py

@@ -4,6 +4,7 @@
 
 from knesset_data_pipelines.committees import background_material_titles
 from knesset_data_pipelines.committees import parsed_document_committee_sessions
+from knesset_data_pipelines.config import AIRFLOW_DEFAULT_EMAILS
 
 
 dag_kwargs = dict(

airflow/dags/dags_generator_etl.py

@@ -14,6 +14,7 @@
 from sqlalchemy.orm import sessionmaker
 
 from knesset_data_pipelines.run_pipeline import list_pipelines, main as run_pipeline
+from knesset_data_pipelines.config import AIRFLOW_DEFAULT_EMAILS
 
 
 dag_kwargs = dict(
@@ -89,14 +90,16 @@ def get_execution_dates(execution_date, external_dag_id, **kwargs):
                 ),
                 random_name_suffix=True,
                 task_id=pipeline_dag_id,
-                dag=dag
+                dag=dag,
+                email=AIRFLOW_DEFAULT_EMAILS,
             )
         else:
             main_task = PythonOperator(
                 python_callable=run_pipeline,
                 task_id=pipeline_id.replace('/', '.'),
                 op_kwargs={'pipeline_id': pipeline_id},
-                dag=dag
+                dag=dag,
+                email=AIRFLOW_DEFAULT_EMAILS,
             )
         for dependency in pipeline_dependencies:
             dependency_dag_id = dependency.replace('/', '.')
@@ -105,5 +108,6 @@ def get_execution_dates(execution_date, external_dag_id, **kwargs):
                 external_dag_id=dependency_dag_id,
                 execution_date_fn=partial(get_execution_dates, external_dag_id=dependency_dag_id),
                 mode='reschedule',
-                dag=dag
+                dag=dag,
+                email=AIRFLOW_DEFAULT_EMAILS,
             ) >> main_task

airflow/dags/google_drive_uploads.py

@@ -3,6 +3,7 @@
 from airflow.operators.python import PythonOperator
 
 from knesset_data_pipelines.google_drive_upload import committee_meeting_protocols
+from knesset_data_pipelines.config import AIRFLOW_DEFAULT_EMAILS
 
 
 dag_kwargs = dict(
@@ -18,5 +19,6 @@
 with DAG('google_drive_uploads.committee_meeting_protocols', **dag_kwargs) as dag:
     PythonOperator(
         python_callable=committee_meeting_protocols.main,
-        task_id='committee_meeting_protocols'
+        task_id='committee_meeting_protocols',
+        email=AIRFLOW_DEFAULT_EMAILS,
     )

airflow/dags/kns_odata.py

@@ -3,6 +3,7 @@
 from airflow.operators.python import PythonOperator
 
 from knesset_data_pipelines import kns_odata
+from knesset_data_pipelines.config import AIRFLOW_DEFAULT_EMAILS
 
 
 dag_kwargs = dict(
@@ -18,5 +19,6 @@
 with DAG('kns_odata', **dag_kwargs) as dag:
     PythonOperator(
         python_callable=kns_odata.compare_parliamentinfo_tables_pipelines,
-        task_id='compare_parliamentinfo_tables_pipelines'
+        task_id='compare_parliamentinfo_tables_pipelines',
+        email=AIRFLOW_DEFAULT_EMAILS,
     )

airflow/dags/members.py

@@ -3,6 +3,7 @@
 from airflow.operators.python import PythonOperator
 
 from knesset_data_pipelines.members_eng import members_eng
+from knesset_data_pipelines.config import AIRFLOW_DEFAULT_EMAILS
 
 
 dag_kwargs = dict(
@@ -18,5 +19,6 @@
 with DAG('members.members_eng', **dag_kwargs) as dag:
     PythonOperator(
         python_callable=members_eng.main,
-        task_id='members_eng'
+        task_id='members_eng',
+        email=AIRFLOW_DEFAULT_EMAILS,
     )

airflow/knesset_data_pipelines/config.py

@@ -54,6 +54,7 @@
 
 DATASERVICE_HTTP_PROXY = os.environ.get('DATASERVICE_HTTP_PROXY')
 
+AIRFLOW_DEFAULT_EMAILS = [e.strip() for e in (os.environ.get('AIRFLOW_DEFAULT_EMAILS') or '').split(',') if e.strip()]
 
 @contextmanager
 def get_google_service_account_json_file_name():