Fixed issue with change tracking order

Author: davidames

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.pythonPath": "${workspaceFolder}\\venv\\Scripts\\python.exe"
+}

modules/DataLoadManager.py

@@ -50,7 +50,6 @@ def start_single_import(self, target_engine, model_name, requested_full_refresh)
 
         data_load_tracker = DataLoadTracker(model_name, json_data, full_refresh, change_tracking_info, self.correlation_id)
 
-
         self.logger.debug(" Change Tracking: this_sync_version: {0} next_sync_version: {1} force_full_load:{2}    : ".format(change_tracking_info.this_sync_version, change_tracking_info.next_sync_version, change_tracking_info.force_full_load()))
         if not full_refresh and change_tracking_info.force_full_load():
             self.logger.info("Change tracking has forced this to be a full load")

modules/DestinationTableManager.py

@@ -24,7 +24,6 @@ def create_schema(self, schema_name):
     def table_exists(self, schema_name, table_name):
         return self.target_engine.dialect.has_table(self.target_engine, table_name, schema_name)
 
-
     def drop_table(self, schema_name, table_name):
         metadata = MetaData()
         self.logger.debug(
@@ -36,9 +35,6 @@ def drop_table(self, schema_name, table_name):
         self.logger.debug(
             "Dropped table {0}.{1}".format(schema_name, table_name))
 
-
-
-
     def create_table(self, schema_name, table_name, columns_configuration, drop_first):
         metadata = MetaData()
 
@@ -63,7 +59,6 @@ def create_table(self, schema_name, table_name, columns_configuration, drop_firs
             self.logger.debug(
                 "Dropped table {0}.{1}".format(schema_name, table_name))
 
-
         self.logger.debug("Creating table {0}.{1}".format(schema_name, table_name))
         table.create(self.target_engine, checkfirst=False)
         self.logger.debug("Created table {0}.{1}".format(schema_name, table_name))
@@ -75,7 +70,6 @@ def create_column(self, configuration):
                       primary_key=configuration.get("primary_key", False),
                       nullable=configuration['nullable'])
 
-
     def rename_table(self, schema_name, source_table_name, target_table_name):
 
         # Steps to efficiently rename a table.
@@ -122,8 +116,9 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         column_list = column_list + ",{0}".format(self.IS_DELETED_COLUMN_NAME)
         column_list = column_list + ",{0}".format(self.CHANGE_VERSION_COLUMN_NAME)
 
-
-        primary_key_column_array = [column_configuration['destination']['name'] for column_configuration in columns_configuration if 'primary_key' in column_configuration['destination'] and column_configuration['destination']['primary_key']]
+        primary_key_column_array = [column_configuration['destination']['name'] for column_configuration in
+                                    columns_configuration if 'primary_key' in column_configuration['destination'] and
+                                    column_configuration['destination']['primary_key']]
 
         primary_key_column_list = ','.join(map(str, primary_key_column_array))
 
@@ -134,15 +129,15 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         sql_builder.write(os.linesep)
         sql_builder.write(" ON CONFLICT({0}) DO UPDATE SET ".format(primary_key_column_list))
 
-        for column_configuratiomn in columns_configuration:
-            sql_builder.write("{0} = EXCLUDED.{0},".format(column_configuratiomn['destination']['name']))
+        for column_configuration in columns_configuration:
+            sql_builder.write("{0} = EXCLUDED.{0},".format(column_configuration['destination']['name']))
             sql_builder.write(os.linesep)
 
         sql_builder.write("{0} = EXCLUDED.{0},".format(self.TIMESTAMP_COLUMN_NAME))
         sql_builder.write(os.linesep)
         sql_builder.write("{0} = EXCLUDED.{0},".format(self.IS_DELETED_COLUMN_NAME))
         sql_builder.write(os.linesep)
-        sql_builder.write("{0} = EXCLUDED.{0},".format(self.CHANGE_VERSION_COLUMN_NAME))
+        sql_builder.write("{0} = EXCLUDED.{0}".format(self.CHANGE_VERSION_COLUMN_NAME))
         sql_builder.write(os.linesep)
 
         self.logger.debug("Upsert executing {0}".format(sql_builder.getvalue()))

modules/data_load_tracking/DataLoadTracker.py

@@ -32,7 +32,7 @@ def start_batch(self):
     def completed_successfully(self):
         self.completed = datetime.now()
         self.total_execution_time = self.completed - self.started
-
+        self.status = "Completed Successfully"
         for batch in self.batches:
             self.total_row_count = self.total_row_count + batch.row_count
 
@@ -91,7 +91,6 @@ def load_completed_successfully(self):
                 self.load_rows_per_second = self.row_count / self.load_execution_time.total_seconds()
 
 
-        # TODO: remove
         def load_skipped_due_to_zero_rows(self):
             self.status = "Skipped - Zero Rows"
             self.load_completed = datetime.now()

modules/data_load_tracking/DataLoadTrackerRepository.py

@@ -1,4 +1,5 @@
 import logging
+from sqlalchemy import desc
 from modules.data_load_tracking.DataLoadExecution import DataLoadExecution, Base
 
 
@@ -13,7 +14,7 @@ def create_tables(self, engine):
 
     def get_last_sync_version(self, model_name):
         session = self.session_maker()
-        result = session.query(DataLoadExecution).filter_by(model_name=model_name, status="Load Completed Successfully").order_by(DataLoadExecution.completed_on).first()
+        result = session.query(DataLoadExecution).filter_by(model_name=model_name, status="Completed Successfully").order_by(desc(DataLoadExecution.completed_on)).first()
 
         if result is None:
             return 0

modules/data_sources/ChangeTrackingInfo.py

@@ -8,6 +8,6 @@ def __init__(self, this_sync_version, next_sync_version):
         self.next_sync_version = next_sync_version
 
     def force_full_load(self):
-        return bool(self.next_sync_version == 0)
+        return bool(self.this_sync_version == 0 or self.next_sync_version == 0)