"""DAG for adjustment file import process.""" from datetime import datetime from airflow import DAG from airflow.operators.python import PythonOperator from lib import constants from tasks.adjustment_file_import.invoke_adjustment_file_import_lambda import \ invoke_adjustment_file_import_lambda_task from tasks.adjustment_file_import.notify_failure import notify_failure_task from tasks.adjustment_file_import.notify_started import notify_started_task from tasks.adjustment_file_import.notify_success import notify_success_task dag = DAG( constants.DAG_ADJUSTMENT_FILE_IMPORT_NAME, description='Import worksheet adjustments', start_date=datetime(2019, 1, 1), default_args={'provide_context': True}, schedule_interval=None ) """ ###################################################################### ############################# OPERATORS ############################## ###################################################################### """ invoke_adjustment_file_import_lambda_operator = PythonOperator( task_id='invoke_adjustment_file_import_lambda', python_callable=invoke_adjustment_file_import_lambda_task, dag=dag ) notify_failure_operator = PythonOperator( dag=dag, python_callable=notify_failure_task, task_id='notify_failure', trigger_rule='one_failed' ) notify_started_operator = PythonOperator( dag=dag, python_callable=notify_started_task, task_id='notify_started' ) notify_success_operator = PythonOperator( dag=dag, python_callable=notify_success_task, task_id='notify_success' ) notify_started_operator >> \ invoke_adjustment_file_import_lambda_operator >> \ [notify_success_operator, notify_failure_operator]