Merge pull request #34 from PageUpPeopleOrg/feature/OSC-1304-log-execution-summary

[OSC-1304] Log execution summary

Author: ChintanRaval

rdl/DataLoadManager.py

@@ -2,6 +2,7 @@
 import uuid
 import logging
 import hashlib
+from datetime import datetime
 from pathlib import Path
 from json import JSONDecodeError
 
@@ -25,6 +26,9 @@ def __init__(self, configuration_path, source_db, target_db, data_load_tracker_r
         self.all_model_pattern = self.model_pattern.format(model_name='*')
 
     def start_imports(self, force_full_refresh_models):
+        self.logger.info(f"Starting Execution ID: '{self.correlation_id}'")
+        execution_start_time = datetime.now()
+
         model_folder = Path(self.configuration_path)
         if not model_folder.is_dir():
             raise NotADirectoryError(self.configuration_path)
@@ -57,6 +61,18 @@ def start_imports(self, force_full_refresh_models):
             self.start_single_import(model_file, request_full_refresh, model_number, total_number_of_models)
 
         self.logger.info("Execution completed.")
+        execution_end_time = datetime.now()
+        total_execution_seconds = int((execution_end_time - execution_start_time).total_seconds())
+        execution_hours = total_execution_seconds // 3600
+        execution_minutes = (total_execution_seconds // 60) % 60
+        execution_seconds = total_execution_seconds % 60
+        total_number_of_rows_processed = self.data_load_tracker_repository.get_execution_rows(self.correlation_id)
+        self.logger.info(
+            f"Completed Execution ID: {self.correlation_id}"
+            f"\n\tModels Processed: {total_number_of_models}"
+            f"\n\tRows Processed: {total_number_of_rows_processed}"
+            f"\n\tExecution Time: {execution_hours}h {execution_minutes}m {execution_seconds}s"
+            f"\n\tAverage of {total_number_of_rows_processed//max(total_execution_seconds, 1)} rows per second.")
 
     def start_single_import(self, model_file, requested_full_refresh, model_number, total_number_of_models):
         model_name = model_file.stem

rdl/data_load_tracking/DataLoadTracker.py

@@ -54,7 +54,7 @@ def data_load_completed(self, execution_status, failure_reason=None):
         self.failure_reason = failure_reason
         for batch in self.batches:
             self.total_row_count += batch.row_count
-        self.rows_per_second = self.total_row_count / (self.total_execution_time.total_seconds() + 1e-10)
+        self.rows_per_second = self.total_row_count / max(self.total_execution_time.total_seconds(), 1)
 
     def get_statistics(self):
         load_type = 'FULL' if self.is_full_refresh else f"INCREMENTAL from " \

rdl/data_load_tracking/DataLoadTrackerRepository.py

@@ -40,6 +40,14 @@ def save(self, data_load_tracker):
         session.commit()
         session.close()
 
+    def get_execution_rows(self, correlation_id):
+        session = self.session_maker()
+        results = session.query(func.sum(DataLoadExecution.rows_processed))\
+            .filter(DataLoadExecution.correlation_id == correlation_id)\
+            .scalar()
+        session.close()
+        return results
+
     def get_full_refresh_since(self, timestamp):
         session = self.session_maker()
         results = session.query(DataLoadExecution.model_name)\

rdl/data_sources/MsSqlDataSource.py

@@ -57,7 +57,7 @@ def create_connection_with_failover(self):
         server = conn_string_data.group('server')
         failover = conn_string_data.group('failover')
         database = conn_string_data.group('database')
-        driver = "{"+conn_string_data.group('driver').replace('+', ' ')+"}"
+        driver = "{" + conn_string_data.group('driver').replace('+', ' ') + "}"
         dsn = f'DRIVER={driver};DATABASE={database};'
 
         username = conn_string_data.group('username')