Update dag

Author: ProgrammingOperative

airflow/dags/postgres_dags/new_workflow.py

@@ -0,0 +1,31 @@
+import json
+import pathlib
+ 
+import airflow
+import requests
+import requests.exceptions as requests_exceptions
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+from airflow.operators.python import PythonOperator
+
+import sys
+import os
+
+sys.path.insert(0, '/home/wacira/10 Academy/week 11/repository/traffic_data_etl/scripts')
+
+dag = DAG(
+   dag_id="Try_dag_out",
+   start_date=airflow.utils.dates.days_ago(14),
+   schedule_interval=None,
+)
+
+def say_hello():
+   return 'HellO Titus'
+
+say_hello = PythonOperator(
+   task_id="Hello_to_me",
+   python_callable = say_hello,
+   dag=dag,
+)
+
+say_hello

airflow/dags/workflow.py

@@ -13,54 +13,48 @@
 
 sys.path.insert(0, '/home/wacira/10 Academy/week 11/repository/traffic_data_etl/scripts')
 
-dag = DAG(
-   dag_id="Try_dag_out",
-   start_date=airflow.utils.dates.days_ago(14),
-   schedule_interval=None,
-)
 
-def hello_world():
-   return 'Hellow Airflow'
+import load_data
+import extract_data
+from load_data import LoadToDB
+from extract_data import ExtractCSV
 
-say_hello = PythonOperator(
-   task_id="say_hellowz",
-   python_callable = hello_world,
-   dag=dag,
-)
-
-say_hello
- 
 
-# import load_data
-# import extract_data
-# from load_data import LoadToDB
-# from extract_data import ExtractCSV
+args={
+   'depends_on_past': False,
+   'email': ['airflow@example.com'],
+   'email_on_failure': False,
+   'email_on_retry': False,
+   'retries': 1,
+   'retry_delay': timedelta(minutes=5)
+    }
 
-# def load_restructure():
-#     return LoadToDB.load_to_db()
+def load_restructure():
+    return LoadToDB.load_to_db()
 
-# def extract():
-#     return ExtractCSV.load_and_restructure()
+def extract():
+    return ExtractCSV.load_and_restructure()
 
-# dag = DAG(
-#    dag_id="Load_Data_files",
-#    start_date=airflow.utils.dates.days_ago(14),
-#    schedule_interval=None,
-# )
+dag = DAG(
+   dag_id="ELT_orchestration",
+   default_args = args,
+   start_date=airflow.utils.dates.days_ago(14),
+   schedule_interval=None,
+)
 
-# extract_data = PythonOperator(
-#    task_id="download_launches",
-#    python_callable = extract,
-#    dag=dag,
-# )
+extract_data = PythonOperator(
+   task_id="download_launches",
+   python_callable = extract,
+   dag=dag,
+)
 
 
-# load_data = PythonOperator(
-#    task_id="load_data",
-#    python_callable= load_restructure,
-#    dag=dag,
-# )
+load_data = PythonOperator(
+   task_id="load_data",
+   python_callable= load_restructure,
+   dag=dag,
+)
 
 
-# extract_data >> load_data
+extract_data >> load_data