Apache Airflow as Job Orchestration

apache_airflow_covid19_create_tasks.py

# Echo task start
task_start = BashOperator(
    task_id = 'start_task',
    bash_command = 'echo start',
    dag = dag
)
# Task 1: scraping daily summary data
summary_scraping = PythonOperator(
    task_id = 'summary_scraping_data',
    python_callable = get_daily_summary,
    dag = dag
)
# Task 2: save the daily summary data
summary_save = PythonOperator(
    task_id = 'summary_save_data',
    python_callable = summary_save_txt,
    provide_context = True,
    dag = dag
)
# Task 3: scraping daily provinces data
provinces_scraping = PythonOperator(
    task_id = 'provinces_scraping_data',
    python_callable = get_daily_summary_provinces,
    dag = dag
)
# Task 4: save the daily summary data
provinces_save = PythonOperator(
    task_id = 'provinces_save_data',
    python_callable = provinces_save_csv,
    provide_context = True,
    dag = dag
)
# Task 5: send norification email
send_email = EmailOperator(
    task_id = 'send_email',
    to = ['[email protected]'],
    subject = 'Covid19 data ',
    html_content = '''
    {date} Covid19 data has been scraped!
    '''.format(date = current_date),
    dag = dag
)
# Task 6: send notification Telegram
send_telegram = PythonOperator(
    task_id = 'send_telegram',
    python_callable = telegram_bot,
    provide_context = True,
    dag = dag
)
# Echo task finish
finish_start = BashOperator(
    task_id = 'finish_task',
    bash_command = 'echo finish',
    dag = dag
)