Add execution info in some log messages

megalista_dataflow/data_sources/big_query/big_query_data_source.py

@@ -153,7 +153,7 @@ def write_transactional_info(self, rows, execution: Execution):
 
             for results in partial_results:
                 for result in results:
-                    logging.get_logger(_LOGGER_NAME).error(result['errors'])
+                    logging.get_logger(_LOGGER_NAME).error(result['errors'], execution=execution)
 
     def _get_now(self):
         return datetime.now().timestamp()

megalista_dataflow/models/execution.py

@@ -14,7 +14,6 @@
 
 from enum import Enum
 from typing import Dict, List, Union, Any
-from config import logging
 from apache_beam.typehints.decorators import with_output_types
 
 class DestinationType(Enum):

megalista_dataflow/third_party/uploaders/appsflyer/appsflyer_s2s_uploader_async.py

@@ -135,7 +135,7 @@ def process(self, batch: Batch, **kwargs):
     if delta_sec < min_duration_sec:
       time.sleep(min_duration_sec - delta_sec)
     logging.get_logger(LOGGER_NAME).info(
-      f"Successfully uploaded {len(success_elements)}/{len(batch.elements)} events.")
+      f"Successfully uploaded {len(success_elements)}/{len(batch.elements)} events.", execution=execution)
 
     execution.successful_records = execution.successful_records + len(success_elements)
     execution.unsuccessful_records = execution.unsuccessful_records + (len(batch.elements) - len(success_elements))

megalista_dataflow/uploaders/campaign_manager/campaign_manager_conversion_uploader.py

@@ -98,7 +98,7 @@ def _do_upload_data(
       'conversions': conversions,
     }
 
-    logger.info(f'Conversions: \n{conversions}')
+    logger.info(f'Conversions: \n{conversions}', execution=execution)
 
     request = service.conversions().batchinsert(
         profileId=campaign_manager_profile_id, body=request_body)
@@ -108,7 +108,7 @@ def _do_upload_data(
     amount_of_failures = 0
 
     if response['hasFailures']:
-      logger.error(f'Error(s) inserting conversions:\n{response}')
+      logger.error(f'Error(s) inserting conversions:\n{response}', execution=execution)
       conversions_status = response['status']
       error_messages = []
 
@@ -121,7 +121,7 @@ def _do_upload_data(
           amount_of_success = amount_of_success + 1
 
       final_error_message = 'Errors from API:\n{}'.format('\n'.join(error_messages))
-      logger.error(final_error_message)
+      logger.error(final_error_message, execution=execution)
       self._add_error(execution, final_error_message)
     else:
       amount_of_success = len(rows)

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_calls_uploader.py

@@ -93,7 +93,7 @@ def process(self, batch: Batch, **kwargs):
     return [batch_with_successful_gclids]
 
   def _do_upload(self, oc_service: Any, execution: Execution, conversion_resource_name: str, customer_id: str, rows: List[Dict[str, Union[str, Dict[str, str]]]]):
-    logging.get_logger(_DEFAULT_LOGGER).info(f'Uploading {len(rows)} offline conversions (calls) on {conversion_resource_name} to Google Ads.')
+    logging.get_logger(_DEFAULT_LOGGER).info(f'Uploading {len(rows)} offline conversions (calls) on {conversion_resource_name} to Google Ads.', execution=execution)
     conversions = [{
           'conversion_action': conversion_resource_name,
           'caller_id': conversion['caller_id'],

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_uploader.py

@@ -100,7 +100,7 @@ def process(self, batch: Batch, **kwargs):
     return [batch_with_successful_gclids]
 
   def _do_upload(self, oc_service, execution, conversion_resource_name, customer_id, rows):
-    logging.get_logger(_DEFAULT_LOGGER).info(f'Uploading {len(rows)} offline conversions on {conversion_resource_name} to Google Ads.')
+    logging.get_logger(_DEFAULT_LOGGER).info(f'Uploading {len(rows)} offline conversions on {conversion_resource_name} to Google Ads.', execution=execution)
     conversions = [{
           'conversion_action': conversion_resource_name,
           'conversion_date_time': utils.format_date(conversion['time']),

megalista_dataflow/uploaders/google_analytics/google_analytics_4_measurement_protocol.py

@@ -81,13 +81,13 @@ def do_process(self, batch: Batch):
       response = requests.post(url,data=json.dumps(payload))
       if response.status_code != 204:
         error_message = f'Error calling GA4 MP {response.status_code}: {str(response.content)}'
-        logging.get_logger(LOGGER_NAME).error(error_message)
+        logging.get_logger(LOGGER_NAME).error(error_message, execution=execution)
         self._add_error(execution, error_message)
       else:
         accepted_elements.append(row)
 
     logging.get_logger(LOGGER_NAME).info(
-      f'Successfully uploaded {len(accepted_elements)}/{len(batch.elements)} events.')
+      f'Successfully uploaded {len(accepted_elements)}/{len(batch.elements)} events.', execution=execution)
 
     execution.successful_records = execution.successful_records + len(accepted_elements)
     execution.unsuccessful_records = execution.unsuccessful_records + (len(batch.elements) - len(accepted_elements))

megalista_dataflow/uploaders/google_analytics/google_analytics_data_import_eraser.py

@@ -84,11 +84,11 @@ def process(self, batch: Batch, **kwargs):
                 return [batch]
             except Exception as e:
                 error_message = f'Error while delete GA Data Import files: {e}'
-                logging.get_logger(LOGGER_NAME).error(error_message)
+                logging.get_logger(LOGGER_NAME).error(error_message, execution=execution)
                 self._add_error(execution, error_message)
         else:
             error_message = f"{data_import_name} - data import not found, please configure it in Google Analytics"
-            logging.get_logger(LOGGER_NAME).error(error_message)
+            logging.get_logger(LOGGER_NAME).error(error_message, execution=execution)
             self._add_error(execution, error_message)
 
     @staticmethod

megalista_dataflow/uploaders/google_analytics/google_analytics_data_import_uploader.py

@@ -104,14 +104,14 @@ def _do_upload_data(self, execution, web_property_id, data_import_name, ga_accou
                                       data_source_id, rows, web_property_id)
             except Exception as e:
                 error_message = f'Error while uploading GA Data: {e}'
-                logging.get_logger(LOGGER_NAME).error(error_message)
+                logging.get_logger(LOGGER_NAME).error(error_message, execution=execution)
                 self._add_error(execution, error_message)
                 execution.unsuccessful_records = execution.unsuccessful_records + len(rows)
             else:
                 execution.successful_records = execution.successful_records + len(rows)
         else:
             error_message = f'{data_import_name} - data import not found, please configure it in Google Analytics'
-            logging.get_logger(LOGGER_NAME).error(error_message)
+            logging.get_logger(LOGGER_NAME).error(error_message, execution=execution)
             self._add_error(execution, error_message)
             execution.unsuccessful_records = execution.unsuccessful_records + len(rows)
 

megalista_dataflow/uploaders/google_analytics/google_analytics_measurement_protocol.py

@@ -89,7 +89,7 @@ def build_hit(self, batch: Batch, row: Dict[str, Any]) -> Dict[str, Any]:
       payload["cu"] = row.get('currency_code')  # Currency code.      
     else:
       error_message = f"Hit type {hit_type} is not supported."
-      logging.get_logger("megalista.GoogleAnalyticsMeasurementProtocolUploader").error(error_message)
+      logging.get_logger("megalista.GoogleAnalyticsMeasurementProtocolUploader").error(error_message, execution=batch.execution)
       self._add_error(batch.execution, error_message)
 
     return payload
@@ -107,7 +107,7 @@ def process(self, batch: Batch, **kwargs):
     response = requests.post(url=self.API_URL, data=payload)
     if response.status_code != 200:
       error_message = f"Error uploading to Analytics HTTP {response.status_code}: {response.raw}"
-      logging.get_logger("megalista.GoogleAnalyticsMeasurementProtocolUploader").error(error_message)
+      logging.get_logger("megalista.GoogleAnalyticsMeasurementProtocolUploader").error(error_message, execution=batch.execution)
       self._add_error(batch.execution, error_message)
       batch.execution.unsuccessful_records = batch.execution.unsuccessful_records + len(rows)
     else:

megalista_dataflow/uploaders/google_analytics/google_analytics_user_list_uploader.py

@@ -163,15 +163,15 @@ def _do_upload_data(self, execution, web_property_id, view_id, data_import_name,
                     media_body=media).execute()
             except Exception as e:
                 error_message = f'Error while uploading GA Data: {e}'
-                logging.get_logger().error(error_message)
+                logging.get_logger().error(error_message, execution=execution)
                 self._add_error(execution, error_message)
                 execution.unsuccessful_records = execution.unsuccessful_records + len(rows)
             else:
                 execution.successful_records = execution.successful_records + len(rows)
 
         else:
             error_message = f"{data_import_name} - data import not found, please configure it in Google Analytics"
-            logging.get_logger().error(error_message)
+            logging.get_logger().error(error_message, execution=execution)
             self._add_error(execution, error_message)
             execution.unsuccessful_records = execution.unsuccessful_records + len(rows)
 

megalista_dataflow/uploaders/utils.py

@@ -72,14 +72,14 @@ def inner(*args, **kwargs):
             if not batch:
                 logger.warning('Skipping upload, received no elements.')
                 return
-            logger.info(f'Uploading {len(batch.elements)} rows...')
+            logger.info(f'Uploading {len(batch.elements)} rows...', execution=batch.execution)
             try:
                 return func(*args, **kwargs)
             except Exception as e:
                 self_._add_error(batch.execution, f'Error uploading data: {e}')
-                logger.error(f'Error uploading data for :{batch.elements}')
-                logger.error(e, exc_info=True)
-                logger.exception('Error uploading data.')
+                logger.error(f'Error uploading data for :{batch.elements}', execution=batch.execution)
+                logger.error(e, exc_info=True, execution=batch.execution)
+                logger.exception('Error uploading data.', execution=batch.execution)
 
         return inner