Updates/cleanup

airqo-platform · Dec 11, 2024 · 0183031 · 0183031
1 parent b82fe30
commit 0183031
Show file tree

Hide file tree

Showing 5 changed files with 8 additions and 8 deletions.
diff --git a/src/workflows/airqo_etl_utils/data_validator.py b/src/workflows/airqo_etl_utils/data_validator.py
@@ -188,8 +188,8 @@ def process_for_big_query(dataframe: pd.DataFrame, table: str) -> pd.DataFrame:
     @staticmethod
     def process_data_for_message_broker(
         data: pd.DataFrame,
-        topic: str,
         caller: str,
+        topic: str = None,
         frequency: Frequency = Frequency.HOURLY,
     ) -> pd.DataFrame:
         """

diff --git a/src/workflows/dags/airnow.py b/src/workflows/dags/airnow.py
@@ -49,8 +49,8 @@ def send_to_message_broker(data: pd.DataFrame, **kwargs):
 
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
         )
         if not data:
             raise AirflowFailException(
@@ -136,8 +136,8 @@ def send_to_message_broker(data: pd.DataFrame, **kwargs):
 
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data:

diff --git a/src/workflows/dags/airqo_bam_measurements.py b/src/workflows/dags/airqo_bam_measurements.py
@@ -150,8 +150,8 @@ def update_latest_data_topic(data: pd.DataFrame, **kwargs):
         )
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.AVERAGED_HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.AVERAGED_HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data:

diff --git a/src/workflows/dags/airqo_measurements.py b/src/workflows/dags/airqo_measurements.py
@@ -110,8 +110,8 @@ def send_hourly_measurements_to_message_broker(
 
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data:
@@ -402,8 +402,8 @@ def send_hourly_measurements_to_message_broker(data: pd.DataFrame, **kwargs):
 
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data:
@@ -453,8 +453,8 @@ def update_latest_data_topic(data: pd.DataFrame, **kwargs):
         )
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.AVERAGED_HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id + unique_str,
+            topic=configuration.AVERAGED_HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data:

diff --git a/src/workflows/dags/kcca_measurements.py b/src/workflows/dags/kcca_measurements.py
@@ -53,8 +53,8 @@ def send_to_message_broker(data: pd.DataFrame, **kwargs):
 
         data = DataValidationUtils.process_data_for_message_broker(
             data=data,
-            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
             caller=kwargs["dag"].dag_id,
+            topic=configuration.HOURLY_MEASUREMENTS_TOPIC,
         )
 
         if not data: