Merge pull request #19 from PageUpPeopleOrg/task/refactor_state_schema

Rename sync versions with correct context

Author: ChintanRaval

modules/BatchDataLoader.py

@@ -1,6 +1,7 @@
 import logging
 from io import StringIO
 from modules.column_transformers.StringTransformers import ToUpper
+from modules.shared import Constants
 
 
 class BatchDataLoader(object):
@@ -23,9 +24,9 @@ def load_batch(self, batch_key_tracker):
         batch_tracker = self.data_load_tracker.start_batch()
 
         self.logger.debug(f"ImportBatch Starting from previous_batch_key: '{batch_key_tracker.bookmarks}'. "
-                          f"Full Refresh: '{self.full_refresh}' "
-                          f"this_sync_version: '{self.change_tracking_info.this_sync_version}'")
-        # TODO ^ this is actually the last_sync_version, log appropriately?
+                          f"Full Refresh: '{self.full_refresh}', "
+                          f"sync_version: '{self.change_tracking_info.sync_version}', "
+                          f"last_sync_version: '{self.change_tracking_info.last_sync_version}'.")
 
         data_frame = self.source_db.get_next_data_frame(self.source_table_config, self.columns,
                                                         self.batch_config, batch_tracker, batch_key_tracker,
@@ -76,8 +77,8 @@ def get_destination_column_name(self, source_column_name):
             if column['source_name'] == source_column_name:
                 return column['destination']['name']
 
-        # Internal columns - map them straight through
-        if source_column_name.startswith("data_pipeline_"):
+        # Audit columns - map them straight through
+        if source_column_name.startswith(Constants.AUDIT_COLUMN_PREFIX):
             return source_column_name
 
         message = f"A source column with name '{source_column_name}' was not found in the column configuration"

modules/DataLoadManager.py

@@ -8,7 +8,7 @@
 from modules.DestinationTableManager import DestinationTableManager
 from modules.data_load_tracking.DataLoadTracker import DataLoadTracker
 from modules.BatchKeyTracker import BatchKeyTracker
-from modules.Shared import Constants
+from modules.shared import Constants
 
 
 class DataLoadManager(object):
@@ -79,7 +79,7 @@ def start_single_import(self, model_file, requested_full_refresh):
             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.next_sync_version
+            last_sync_version = last_successful_data_load_execution.sync_version
 
         destination_table_manager = DestinationTableManager(self.target_db)
         change_tracking_info = self.source_db.init_change_tracking(model_config['source_table'], last_sync_version)

modules/DestinationTableManager.py

@@ -6,7 +6,7 @@
 from sqlalchemy import MetaData, DateTime, Boolean, BigInteger
 from sqlalchemy.schema import Column, Table
 from sqlalchemy.sql import func
-from modules.Shared import Constants
+from modules.shared import Constants
 
 
 class DestinationTableManager(object):
@@ -100,39 +100,35 @@ def rename_table(self, schema_name, source_table_name, target_table_name):
         self.logger.debug(f"Table Rename, executing '{sql}'")
         self.target_db.execute(sql)
 
-    def upsert_table(self, schema_name, source_table_name, target_table_name, columns_configuration):
-        column_array = list(map(lambda column: column['destination']['name'], columns_configuration))
+    def upsert_table(self, schema_name, source_table_name, target_table_name, columns_config):
+        column_array = list(map(lambda column: column['destination']['name'], columns_config))
         column_list = ','.join(map(str, column_array))
         column_list = column_list + f",{Constants.AuditColumnNames.TIMESTAMP}"
         column_list = column_list + f",{Constants.AuditColumnNames.IS_DELETED}"
         column_list = column_list + f",{Constants.AuditColumnNames.CHANGE_VERSION}"
 
-        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_config['destination']['name'] for column_config in
+                                    columns_config if 'primary_key' in column_config['destination'] and
+                                    column_config['destination']['primary_key']]
 
         primary_key_column_list = ','.join(map(str, primary_key_column_array))
 
         sql_builder = io.StringIO()
-        sql_builder.write(f"INSERT INTO {schema_name}.{target_table_name} ({column_list})")
-        sql_builder.write(os.linesep)
-        sql_builder.write(f" SELECT {column_list} FROM {schema_name}.{source_table_name}")
-        sql_builder.write(os.linesep)
+        sql_builder.write(f"INSERT INTO {schema_name}.{target_table_name} ({column_list}) \n")
+        sql_builder.write(f" SELECT {column_list} FROM {schema_name}.{source_table_name} \n")
         sql_builder.write(f" ON CONFLICT({primary_key_column_list}) DO UPDATE SET ")
 
-        for column_configuration in columns_configuration:
-            sql_builder.write("{0} = EXCLUDED.{0},".format(column_configuration['destination']['name']))
-            sql_builder.write(os.linesep)
+        for column_config in columns_config:
+            sql_builder.write("{0} = EXCLUDED.{0},\n".format(column_config['destination']['name']))
 
-        sql_builder.write("{0} = EXCLUDED.{0},".format(Constants.AuditColumnNames.TIMESTAMP))
-        sql_builder.write(os.linesep)
-        sql_builder.write("{0} = EXCLUDED.{0},".format(Constants.AuditColumnNames.IS_DELETED))
-        sql_builder.write(os.linesep)
-        sql_builder.write("{0} = EXCLUDED.{0}".format(Constants.AuditColumnNames.CHANGE_VERSION))
-        sql_builder.write(os.linesep)
+        sql_builder.write("{0} = EXCLUDED.{0},\n".format(Constants.AuditColumnNames.TIMESTAMP))
+        sql_builder.write("{0} = EXCLUDED.{0},\n".format(Constants.AuditColumnNames.IS_DELETED))
+        sql_builder.write("{0} = EXCLUDED.{0};\n".format(Constants.AuditColumnNames.CHANGE_VERSION))
 
-        self.logger.debug(f"UPSERT executing '{sql_builder.getvalue()}'")
-        self.target_db.execute(sql_builder.getvalue())
+        upsert_sql = sql_builder.getvalue()
+
+        self.logger.debug(f"UPSERT executing '{upsert_sql}'")
+        self.target_db.execute(upsert_sql)
         self.logger.debug("UPSERT completed")
 
         sql_builder.close()

modules/RelationalDataLoader.py

@@ -2,7 +2,7 @@
 import argparse
 from sqlalchemy import create_engine
 from modules.DataLoadManager import DataLoadManager
-from modules.Shared import Constants
+from modules.shared import Constants
 from modules.data_load_tracking.DataLoadTrackerRepository import DataLoadTrackerRepository
 from modules.data_sources.DataSourceFactory import DataSourceFactory
 from sqlalchemy.orm import sessionmaker

modules/Shared.py

@@ -1,22 +1,24 @@
 from sqlalchemy import Column, DateTime, Integer, String, Boolean, BigInteger
 from sqlalchemy.sql import func
 from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.dialects.postgresql import UUID
+from modules.shared import Constants
 
 Base = declarative_base()
 
 
 class DataLoadExecution(Base):
     __tablename__ = 'data_load_execution'
-    __table_args__ = {'schema': 'data_pipeline'}
+    __table_args__ = {'schema': f'{Constants.DATA_PIPELINE_EXECUTION_SCHEMA_NAME}'}
     id = Column(Integer, primary_key=True)
+    correlation_id = Column(UUID(as_uuid=True), nullable=True)
     model_name = Column(String(250), nullable=False)
+    status = Column(String(25), nullable=False)
+    last_sync_version = Column(BigInteger, nullable=False)
+    sync_version = Column(BigInteger, nullable=False)
     is_full_refresh = Column(Boolean, nullable=False)
-    this_sync_version = Column(BigInteger, nullable=False)
-    next_sync_version = Column(BigInteger, nullable=False)
+    full_refresh_reason = Column(String(100), nullable=False)
     completed_on = Column(DateTime(timezone=True), server_default=func.now())
     execution_time_ms = Column(Integer, nullable=False)
     rows_processed = Column(Integer, nullable=False)
-    correlation_id = Column(String(250), nullable=True)
-    status = Column(String(25), nullable=False)
     model_checksum = Column(String(100), nullable=False)
-    full_refresh_reason = Column(String(100), nullable=False)

modules/data_load_tracking/DataLoadExecution.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from modules.Shared import Constants
+from modules.shared import Constants
 
 
 class DataLoadTracker:
@@ -51,8 +51,8 @@ def completed_successfully(self):
 
     def get_statistics(self):
         load_type = 'FULL' if self.is_full_refresh else f"INCREMENTAL from " \
-                                                        f"version '{self.change_tracking_info.this_sync_version}' " \
-                                                        f"to '{self.change_tracking_info.next_sync_version}'"
+                                                        f"version '{self.change_tracking_info.last_sync_version}' " \
+                                                        f"to '{self.change_tracking_info.sync_version}'"
         return f"Rows: {self.total_row_count}," \
                f"Load type: {load_type}, " \
                f"Total Execution Time: {self.total_execution_time} @ {self.rows_per_second:.2f} rows per second "

modules/data_load_tracking/DataLoadTracker.py

@@ -1,7 +1,7 @@
 import logging
 from sqlalchemy import desc
 from modules.data_load_tracking.DataLoadExecution import DataLoadExecution, Base
-from modules.Shared import Constants
+from modules.shared import Constants
 
 
 class DataLoadTrackerRepository(object):
@@ -15,26 +15,23 @@ def ensure_schema_exists(self, engine):
 
     def get_last_successful_data_load_execution(self, model_name):
         session = self.session_maker()
-        return session.query(DataLoadExecution).filter_by(
-            model_name=model_name,
-            status=Constants.ExecutionStatus.COMPLETED_SUCCESSFULLY).order_by(
-            desc(
-                DataLoadExecution.completed_on)).first()
+        return session.query(DataLoadExecution)\
+            .filter_by(model_name=model_name, status=Constants.ExecutionStatus.COMPLETED_SUCCESSFULLY)\
+            .order_by(desc(DataLoadExecution.completed_on))\
+            .first()
 
     def save(self, data_load_tracker):
-
         data_load_execution = DataLoadExecution(
-            model_name=data_load_tracker.model_name,
             correlation_id=data_load_tracker.correlation_id,
+            model_name=data_load_tracker.model_name,
+            status=data_load_tracker.status,
+            last_sync_version=data_load_tracker.change_tracking_info.last_sync_version,
+            sync_version=data_load_tracker.change_tracking_info.sync_version,
             is_full_refresh=data_load_tracker.is_full_refresh,
-            this_sync_version=data_load_tracker.change_tracking_info.this_sync_version,
-            next_sync_version=data_load_tracker.change_tracking_info.next_sync_version,
-            execution_time_ms=int(
-                data_load_tracker.total_execution_time.total_seconds() * 1000),
+            full_refresh_reason=data_load_tracker.full_refresh_reason,
+            execution_time_ms=int(data_load_tracker.total_execution_time.total_seconds() * 1000),
             rows_processed=data_load_tracker.total_row_count,
-            status=data_load_tracker.status,
-            model_checksum=data_load_tracker.model_checksum,
-            full_refresh_reason=data_load_tracker.full_refresh_reason)
+            model_checksum=data_load_tracker.model_checksum)
 
         session = self.session_maker()
         session.add(data_load_execution)

modules/data_load_tracking/DataLoadTrackerRepository.py

@@ -1,10 +1,5 @@
-
 class ChangeTrackingInfo:
-    this_sync_version = 0  # TODO rename `this_sync_version` to `last_sync_version`
-    next_sync_version = 0  # TODO rename `next_sync_version` to `this?current?new?_sync_version`
-    force_full_load = 0
-
-    def __init__(self, this_sync_version, next_sync_version, force_full_load):
-        self.this_sync_version = this_sync_version
-        self.next_sync_version = next_sync_version
+    def __init__(self, last_sync_version, sync_version, force_full_load):
+        self.last_sync_version = last_sync_version
+        self.sync_version = sync_version
         self.force_full_load = force_full_load