Apache Airflow as Job Orchestration

apache_airflow_covid19_save_summary.py

# Function to save the daily aggregated data
def summary_save_txt(**context):
    value = context['task_instance'].xcom_pull(task_ids = 'summary_scraping_data')
    with open(dir_path+'/data/covid19/summary_covid19.txt','r') as f:
        lines = f.read().splitlines()
        last_line = lines[-1]
        if last_line == value:
            notif = 'Last update:',re.findall(r'^(.+?),',last_line)[0]
        else:
            with open(dir_path+'/data/covid19/summary_covid19.txt','a+') as ff:
                ff.write('{}\n'.format(value))
                ff.close()
            notif = 'Up to date data:', re.findall(r'^(.+?),',value)[0]
    return(notif)