Merge pull request #53 from pageuppeople-opensource/feature/OSC-1458-reduce-api-calls-to-datasource

OSC-1458 - Reduce api calls to FooDataSource

Author: ChintanRaval

rdl/DataLoadManager.py

@@ -85,17 +85,29 @@ def start_single_import(self, model_file, requested_full_refresh, model_number,
             self.logger.error(f"Failed to read model file '{model_file_full_path}' with error: '{str(exception)}'")
             raise exception
 
-        self.source_db.assert_data_source_is_valid(model_config['source_table'], model_config['columns'])
-
         last_sync_version = 0
         last_successful_data_load_execution = \
             self.data_load_tracker_repository.get_last_successful_data_load_execution(model_name)
 
         if last_successful_data_load_execution is not None:
             last_sync_version = last_successful_data_load_execution.sync_version
 
+        source_table_info = self.source_db.get_table_info(model_config['source_table'], last_sync_version)
+        config_source_column_names = list(map(lambda col_config: col_config['source_name'], model_config['columns']))
+        invalid_config_source_column_names = []
+
+        for col_name in config_source_column_names:
+            if col_name not in source_table_info.column_names:
+                invalid_config_source_column_names.append(col_name)
+
+        if invalid_config_source_column_names:
+            message = "Column(s) {column_names} not found in source table {source_table_name}".format(
+                column_names=", ".join(invalid_config_source_column_names),
+                source_table_name=model_config['source_table']['name'])
+            raise ValueError(message)
+
         destination_table_manager = DestinationTableManager(self.target_db)
-        change_tracking_info = self.source_db.get_change_tracking_info(model_config['source_table'], last_sync_version)
+        change_tracking_info = source_table_info.change_tracking_info
 
         last_successful_execution_exists = last_successful_data_load_execution is not None
         model_changed = (not last_successful_execution_exists) or \

rdl/data_sources/MsSqlDataSource.py

@@ -12,6 +12,7 @@
 
 from rdl.ColumnTypeResolver import ColumnTypeResolver
 from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
+from rdl.data_sources.SourceTableInfo import SourceTableInfo
 from rdl.shared import Providers
 from rdl.shared.Utils import prevent_senstive_data_logging
 
@@ -111,21 +112,11 @@ def __build_select_statement(self, table_config, columns, batch_config, batch_ke
 
         return f"{select_sql} \n {from_sql} \n {where_sql} \n {order_by_sql};"
 
-    # Returns an array of configured_columns containing only columns that this data source supports. Logs invalid ones.
-    def assert_data_source_is_valid(self, table_config, configured_columns):
+    def get_table_info(self, table_config, last_known_sync_version):
         columns_in_database = self.__get_table_columns(table_config)
-
-        for column in configured_columns:
-            self.__assert_column_exists(column['source_name'],
-                                        columns_in_database,
-                                        f"{table_config['schema']}.{table_config['name']}")
-
-    def __assert_column_exists(self, column_name, columns_in_database, table_name):
-        if column_name in columns_in_database:
-            return True
-
-        message = f'Column {column_name} does not exist in source table {table_name}'
-        raise ValueError(message)
+        change_tracking_info = self.__get_change_tracking_info(table_config, last_known_sync_version)
+        source_table_info = SourceTableInfo(columns_in_database, change_tracking_info)
+        return source_table_info
 
     def __get_table_columns(self, table_config):
         metadata = MetaData()
@@ -149,7 +140,7 @@ def get_next_data_frame(self, table_config, columns, batch_config, batch_tracker
 
         return data_frame
 
-    def get_change_tracking_info(self, table_config, last_known_sync_version):
+    def __get_change_tracking_info(self, table_config, last_known_sync_version):
 
         if last_known_sync_version is None:
             last_known_sync_version = 'NULL'

rdl/data_sources/SourceTableInfo.py

@@ -0,0 +1,7 @@
+from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
+
+
+class SourceTableInfo:
+    def __init__(self, column_names: [], change_tracking_info: ChangeTrackingInfo):
+        self.column_names = column_names
+        self.change_tracking_info = change_tracking_info

tests/unit_tests/test_MsSqlDataSource.py

@@ -59,20 +59,20 @@ def tearDownClass(cls):
         TestMsSqlDataSource.data_source.database_engine.execute(text(TEAR_DOWN_STRING))
         TestMsSqlDataSource.data_source = None
 
-    def test_get_change_tracking_info(self):
+    def test_get_table_info_change_tracking_info(self):
 
         last_sync_version = None
         for table in TestMsSqlDataSource.table_configs:
             print("TESTING ON TABLE: " + table["source_table"]["name"])
             print("FIRST TEST: INITIALISE TABLE")
-            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
-                table["source_table"], last_sync_version)
+            results = TestMsSqlDataSource.data_source.get_table_info(
+                table["source_table"], last_sync_version).change_tracking_info
             self.assertEqual(results.force_full_load, True)
 
             print("SECOND TEST: NO CHANGES")
             last_sync_version = results.sync_version
-            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
-                table["source_table"], last_sync_version)
+            results = TestMsSqlDataSource.data_source.get_table_info(
+                table["source_table"], last_sync_version).change_tracking_info
             self.assertEqual(results.force_full_load, False)
 
             print("OPERATION TESTS")
@@ -82,20 +82,22 @@ def test_get_change_tracking_info(self):
                 TestMsSqlDataSource.data_source.database_engine.execute(
                     text(operation_string).execution_options(autocommit=True))
 
-                results = TestMsSqlDataSource.data_source.get_change_tracking_info(
-                    table["source_table"], last_sync_version)
+                results = TestMsSqlDataSource.data_source.get_table_info(
+                    table["source_table"], last_sync_version).change_tracking_info
                 self.assertEqual(results.force_full_load, False, msg="Failed on: " + operation_string)
                 last_sync_version = results.sync_version
 
             print("EXTRA TEST: NO CHANGES")
             last_sync_version = results.sync_version
-            results = TestMsSqlDataSource.data_source.get_change_tracking_info(
-                table["source_table"], last_sync_version)
+            results = TestMsSqlDataSource.data_source.get_table_info(
+                table["source_table"], last_sync_version).change_tracking_info
             self.assertEqual(results.force_full_load, False)
 
             print("EXTRA TEST: LOST TRACK")
             last_sync_version = -1
-            results = TestMsSqlDataSource.data_source.get_change_tracking_info(table["source_table"], last_sync_version)
+            results = TestMsSqlDataSource.data_source\
+                .get_table_info(table["source_table"], last_sync_version)\
+                .change_tracking_info
             self.assertEqual(results.force_full_load, True)
 
     def test_can_handle_connection_string(self):