chore: update big query ingestion logs (#1210)

Author: davidgamez

functions-python/big_query_ingestion/src/common/bq_data_transfer.py

@@ -29,12 +29,12 @@ def create_bigquery_dataset(self):
         dataset_ref = bigquery.DatasetReference(project_id, dataset_id)
         try:
             self.bigquery_client.get_dataset(dataset_ref)
-            logging.info(f"Dataset {dataset_id} already exists.")
+            logging.info("Dataset %s already exists.", dataset_id)
         except Exception:
             dataset = bigquery.Dataset(dataset_ref)
             dataset.location = dataset_location
             self.bigquery_client.create_dataset(dataset)
-            logging.info(f"Created dataset {dataset_id}.")
+            logging.info("Created dataset %s", dataset_id)
 
     def create_bigquery_table(self):
         """Creates a BigQuery table if it does not exist."""
@@ -43,7 +43,7 @@ def create_bigquery_table(self):
 
         try:
             self.bigquery_client.get_table(table_ref)
-            logging.info(f"Table {table_id} already exists.")
+            logging.info("Table %s  already exists.", table_id)
         except Exception:
             if self.schema_path is None:
                 raise Exception("Schema path is not provided")
@@ -53,7 +53,10 @@ def create_bigquery_table(self):
             table = bigquery.Table(table_ref, schema=schema)
             table = self.bigquery_client.create_table(table)
             logging.info(
-                f"Created table {table.project}.{table.dataset_id}.{table.table_id}"
+                "Created table %s.%s.%s",
+                table.project,
+                table.dataset_id,
+                table.table_id,
             )
 
     def load_data_to_bigquery(self):
@@ -68,7 +71,7 @@ def load_data_to_bigquery(self):
         for blob in blobs:
             uri = f"gs://{bucket_name}/{blob.name}"
             source_uris.append(uri)
-        logging.info(f"Found {len(source_uris)} files to load to BigQuery.")
+        logging.info("Found %s files to load to BigQuery.", len(source_uris))
 
         if len(source_uris) > 0:
             # Load the data to BigQuery
@@ -82,29 +85,35 @@ def load_data_to_bigquery(self):
             try:
                 load_job.result()  # Wait for the job to complete
                 logging.info(
-                    f"Loaded {len(source_uris)} files into "
-                    f"{table_ref.project}.{table_ref.dataset_id}.{table_ref.table_id}"
+                    "Loaded %s files into %s.%s.%s.",
+                    len(source_uris),
+                    table_ref.project,
+                    table_ref.dataset_id,
+                    table_ref.table_id,
                 )
                 # If successful, delete the blobs
                 for blob in blobs:
                     blob.delete()
-                    logging.info(f"Deleted blob: {blob.name}")
+                    logging.debug("Deleted blob: %s", blob.name)
+                logging.info("Deleted blobs")
             except Exception as e:
-                logging.error(f"An error occurred while loading data to BigQuery: {e}")
+                logging.error("An error occurred while loading data to BigQuery: %s", e)
                 for error in load_job.errors:
-                    logging.error(f"Error: {error['message']}")
+                    logging.error("Error: %s", error["message"])
                     if "location" in error:
-                        logging.error(f"Location: {error['location']}")
+                        logging.error("Location: %s", error["location"])
                     if "reason" in error:
-                        logging.error(f"Reason: {error['reason']}")
+                        logging.error("Reason: %s", error["reason"])
 
     def send_data_to_bigquery(self):
         """Full process to send data to BigQuery."""
         try:
             self.create_bigquery_dataset()
             self.create_bigquery_table()
             self.load_data_to_bigquery()
-            return "Data successfully loaded to BigQuery", 200
+            msg = "Data successfully loaded to BigQuery"
+            logging.info(msg)
+            return msg, 200
         except Exception as e:
-            logging.error(f"An error occurred: {e}")
+            logging.error("An error occurred: %s", e)
             return f"Error while loading data: {e}", 500

functions-python/big_query_ingestion/src/main.py

@@ -2,24 +2,22 @@
 
 import functions_framework
 
-from shared.helpers.logger import Logger
+from shared.helpers.logger import init_logger
 from gbfs.gbfs_big_query_ingest import BigQueryDataTransferGBFS
 from gtfs.gtfs_big_query_ingest import BigQueryDataTransferGTFS
 
-logging.basicConfig(level=logging.INFO)
+init_logger()
 
 
 @functions_framework.http
 def ingest_data_to_big_query_gtfs(_):
     """Google Storage to Big Query data ingestion for GTFS data"""
-    Logger.init_logger()
     logging.info("Function triggered")
     return BigQueryDataTransferGTFS().send_data_to_bigquery()
 
 
 @functions_framework.http
 def ingest_data_to_big_query_gbfs(_):
     """Google Storage to Big Query data ingestion for GBFS data"""
-    Logger.init_logger()
     logging.info("Function triggered")
     return BigQueryDataTransferGBFS().send_data_to_bigquery()

functions-python/big_query_ingestion/tests/test_main.py

@@ -9,11 +9,7 @@
 
 class TestMain(unittest.TestCase):
     @patch("main.BigQueryDataTransferGTFS")
-    @patch("shared.helpers.logger.Logger.init_logger")
-    @patch("main.logging.info")
-    def test_ingest_data_to_big_query_gtfs(
-        self, mock_logging_info, mock_init_logger, mock_big_query_transfer_gtfs
-    ):
+    def test_ingest_data_to_big_query_gtfs(self, mock_big_query_transfer_gtfs):
         mock_instance = mock_big_query_transfer_gtfs.return_value
         mock_instance.send_data_to_bigquery.return_value = (
             "Data successfully loaded to BigQuery",
@@ -22,17 +18,11 @@ def test_ingest_data_to_big_query_gtfs(
 
         response = ingest_data_to_big_query_gtfs(None)
 
-        mock_init_logger.assert_called_once()
-        mock_logging_info.assert_any_call("Function triggered")
         mock_instance.send_data_to_bigquery.assert_called_once()
         self.assertEqual(response, ("Data successfully loaded to BigQuery", 200))
 
     @patch("main.BigQueryDataTransferGBFS")
-    @patch("shared.helpers.logger.Logger.init_logger")
-    @patch("main.logging.info")
-    def test_ingest_data_to_big_query_gbfs(
-        self, mock_logging_info, mock_init_logger, mock_biq_query_transfer_gbfs
-    ):
+    def test_ingest_data_to_big_query_gbfs(self, mock_biq_query_transfer_gbfs):
         mock_instance = mock_biq_query_transfer_gbfs.return_value
         mock_instance.send_data_to_bigquery.return_value = (
             "Data successfully loaded to BigQuery",
@@ -41,7 +31,5 @@ def test_ingest_data_to_big_query_gbfs(
 
         response = ingest_data_to_big_query_gbfs(None)
 
-        mock_init_logger.assert_called_once()
-        mock_logging_info.assert_any_call("Function triggered")
         mock_instance.send_data_to_bigquery.assert_called_once()
         self.assertEqual(response, ("Data successfully loaded to BigQuery", 200))