chore: update pre-processing analytics logs (#1224)

Author: davidgamez

functions-python/preprocessed_analytics/requirements.txt

@@ -17,6 +17,7 @@ geoalchemy2==0.14.7
 # Google specific packages for this function
 google-cloud-bigquery
 google-cloud-storage
+flask
 
 # Additional packages for this function
 pandas

functions-python/preprocessed_analytics/src/main.py

@@ -6,12 +6,12 @@
 import functions_framework
 from flask import Response
 
-from shared.helpers.logger import Logger
+from shared.helpers.logger import init_logger
 from processors.base_analytics_processor import NoFeedDataException
 from processors.gbfs_analytics_processor import GBFSAnalyticsProcessor
 from processors.gtfs_analytics_processor import GTFSAnalyticsProcessor
 
-logging.basicConfig(level=logging.INFO)
+init_logger()
 
 
 def get_compute_date(request: flask.Request) -> datetime:
@@ -33,29 +33,29 @@ def preprocess_analytics(request: flask.Request, processor_class) -> Response:
     """
     Common logic to process analytics using the given processor class.
     """
-    Logger.init_logger()
-    logging.info(f"{processor_class.__name__} Function triggered")
+    logging.info("Function triggered: %s", processor_class.__name__)
     compute_date = get_compute_date(request)
-    logging.info(f"Compute date: {compute_date}")
+    logging.info("Compute date: %s", compute_date)
     try:
         processor = processor_class(compute_date)
         processor.run()
     except NoFeedDataException as e:
-        logging.warning(f"No feed data found for date {compute_date}: {e}")
+        logging.warning("No feed data found for date %s: %s", compute_date, e)
         return Response(f"No feed data found for date {compute_date}: {e}", status=404)
     except Exception as e:
         # Extracting the traceback details
         tb = traceback.format_exc()
+        logging.error("Error processing %s analytics: %s", processor_class.__name__, e)
         logging.error(
-            f"Error processing {processor_class.__name__} analytics: {e}\nTraceback:\n{tb}"
+            "Error trace processing %s analytics: %s", processor_class.__name__, tb
         )
         return Response(
             f"Error processing analytics for date {compute_date}: {e}", status=500
         )
 
-    return Response(
-        f"Successfully processed analytics for date: {compute_date}", status=200
-    )
+    message = f"Successfully processed analytics for date: {compute_date}"
+    logging.info(message)
+    return Response(message, status=200)
 
 
 @functions_framework.http

functions-python/preprocessed_analytics/src/processors/base_analytics_processor.py

@@ -61,7 +61,7 @@ def _load_json(self, file_name: str) -> Tuple[List[Dict], storage.Blob]:
                 )
             except Exception as e:
                 logging.warning(
-                    f"Unable to convert data to DataFrame using Pandas: {e}"
+                    "Unable to convert data to DataFrame using Pandas: %s", e
                 )
                 return json.loads(json_data), blob
         return [], blob
@@ -72,13 +72,13 @@ def _save_blob(blob: storage.Blob, data: List[Dict]) -> None:
             # Convert the data to JSON format
             json_data = pd.DataFrame(data).to_json(orient="records", date_format="iso")
         except Exception as e:
-            logging.warning(f"Unable to convert data to JSON using Pandas: {e}")
+            logging.warning("Unable to convert data to JSON using Pandas: %s", e)
             json_data = json.dumps(data, default=str)
 
         # Save the JSON file to the specified GCS bucket
         blob.upload_from_string(json_data, content_type="application/json")
         blob.make_public()
-        logging.info(f"{blob.name} saved to bucket")
+        logging.info("%s saved to bucket", blob.name)
 
     def _save_json(self, file_name: str, data: List[Dict]) -> None:
         # Save the JSON file to the specified GCS bucket
@@ -90,7 +90,7 @@ def aggregate_summary_files(
     ) -> None:
         blobs = self.analytics_bucket.list_blobs(prefix="summary/summary_")
         for blob in blobs:
-            logging.info(f"Aggregating data from {blob.name}")
+            logging.info("Aggregating data from %s", blob.name)
             summary_data, _ = self._load_json(blob.name)
             for key, new_data in summary_data.items():
                 if key in metrics_file_data:
@@ -129,7 +129,7 @@ def save_analytics(self) -> None:
         file_name = f"analytics_{self.run_date.strftime('%Y-%m-%d')}.json"
         self._save_json(file_name, self.data)
         self.save()
-        logging.info(f"Analytics saved to bucket as {file_name}")
+        logging.info("Analytics saved to bucket as %s", file_name)
 
     @with_db_session
     def run(self, db_session: Session) -> None:
@@ -139,16 +139,16 @@ def run(self, db_session: Session) -> None:
         self.save_summary()
         self.save_analytics()
         self.update_analytics_files()
-        logging.info(f"Finished running analytics for date: {self.run_date}")
+        logging.info("Finished running analytics for date: %s", self.run_date)
 
     def _get_data(self, db_session: Session):
         query = self.get_latest_data(db_session)
         all_results = query.all()
         if len(all_results) == 0:
             raise NoFeedDataException("No feed data found")
-        logging.info(f"Loaded {len(all_results)} feeds to process")
+        logging.info("Loaded %s feeds to process", len(all_results))
         unique_feeds = {result[0].stable_id: result for result in all_results}
-        logging.info(f"Nb of unique feeds loaded: {len(unique_feeds)}")
+        logging.info("Nb of unique feeds loaded: %s", len(unique_feeds))
         return [(result[0], result[1]) for result in unique_feeds.values()]
 
     def update_analytics_files(self) -> None:
@@ -189,4 +189,4 @@ def update_analytics_files(self) -> None:
             )
 
         except Exception as e:
-            logging.error(f"Error updating analytics files: {e}")
+            logging.error("Error updating analytics files: %s", e)

functions-python/preprocessed_analytics/tests/test_main.py

@@ -58,24 +58,21 @@ def test_process_analytics_gbfs_success(self, mock_process_analytics, mock_run):
         )
         self.assertEqual(response.status_code, 200)
 
-    @patch("main.Logger.init_logger")
     @patch("main.GTFSAnalyticsProcessor.run")
-    def test_process_analytics_gtfs_error(self, mock_run, _):
+    def test_process_analytics_gtfs_error(self, mock_run):
         mock_run.side_effect = Exception("Test error")
         response = preprocess_analytics_gtfs(self.mock_request)
         self.assertEqual(response.status_code, 500)
 
-    @patch("main.Logger.init_logger")
     @patch("main.GBFSAnalyticsProcessor.run")
-    def test_process_analytics_gbfs_error(self, mock_run, _):
+    def test_process_analytics_gbfs_error(self, mock_run):
         mock_run.side_effect = Exception("Test error")
         response = preprocess_analytics_gbfs(self.mock_request)
         self.assertEqual(response.status_code, 500)
 
-    @patch("main.Logger.init_logger")
     @patch("main.GTFSAnalyticsProcessor.run")
     @patch("main.GTFSAnalyticsProcessor.__init__")
-    def test_process_analytics_success(self, mock_init, mock_run, _):
+    def test_process_analytics_success(self, mock_init, mock_run):
         mock_run.return_value = None
         mock_init.return_value = None
         response = preprocess_analytics(self.mock_request, GTFSAnalyticsProcessor)
@@ -84,9 +81,8 @@ def test_process_analytics_success(self, mock_init, mock_run, _):
             "Successfully processed analytics for date:", response.data.decode()
         )
 
-    @patch("main.Logger.init_logger")
     @patch("main.GTFSAnalyticsProcessor.run")
-    def test_process_analytics_failure(self, mock_run, _):
+    def test_process_analytics_failure(self, mock_run):
         mock_run.side_effect = Exception("Processing error")
         response = preprocess_analytics(self.mock_request, GTFSAnalyticsProcessor)
         self.assertEqual(response.status_code, 500)