[OSC-1294] give an audit column prefix as defined by an argument (#31)

* give an audit column prefix as defined by an argument

* update the audit column prefix source of truth

* update default prefix from rdl -> rdl_

Author: seanbudd

README.md

@@ -9,9 +9,10 @@ A utility for taking data from MS-SQL and loading it into PostgreSQL
 `py -m rdl --help`
 
 ```text
-usage: py -m rdl process [-h] [-f [FORCE_FULL_REFRESH_MODELS]] [-l [LOG_LEVEL]]
-                      source-connection-string destination-connection-string
-                      configuration-folder
+usage: py -m rdl process [-h] [-f [FORCE_FULL_REFRESH_MODELS]]
+                           [-l [LOG_LEVEL]] [-p [AUDIT_COLUMN_PREFIX]]
+                           source-connection-string
+                           destination-connection-string configuration-folder
 
 positional arguments:
   source-connection-string
@@ -38,9 +39,13 @@ optional arguments:
   -l [LOG_LEVEL], --log-level [LOG_LEVEL]
                         Set the logging output level. ['CRITICAL', 'ERROR',
                         'WARNING', 'INFO', 'DEBUG']
+  -p [AUDIT_COLUMN_PREFIX], --audit-column-prefix [AUDIT_COLUMN_PREFIX]
+                        Set the audit column prefix, used in the destination
+                        schema. Default is 'rdl_'.
 
-usage: py -m rdl audit [-h] [-l [LOG_LEVEL]]
-                    destination-connection-string model-type timestamp
+
+usage: py -m rdl audit [-h] [-l [LOG_LEVEL]] [-p [AUDIT_COLUMN_PREFIX]]
+                         destination-connection-string model-type timestamp
 
 positional arguments:
   destination-connection-string
@@ -50,15 +55,19 @@ positional arguments:
   model-type            Use the command FULL to return full refresh models or
                         the command INCR to return only the incremental models
                         since the timestamp
-  timestamp             ISO 8601 datetime with timezone (`yyyy-mm-ddThh:mm:ss.nnnnnn+|-hh:mm`) used to provide information on all
-                        actions since the specified date. Eg
-                        '2019-02-14T01:55:54.123456+00:00'.
+  timestamp             ISO 8601 datetime with timezone (yyyy-mm-
+                        ddThh:mm:ss.nnnnnn+|-hh:mm) used to provide
+                        information on all actions since the specified date.
+                        Eg '2019-02-14T01:55:54.123456+00:00'.
 
 optional arguments:
   -h, --help            show this help message and exit
   -l [LOG_LEVEL], --log-level [LOG_LEVEL]
                         Set the logging output level. ['CRITICAL', 'ERROR',
                         'WARNING', 'INFO', 'DEBUG']
+  -p [AUDIT_COLUMN_PREFIX], --audit-column-prefix [AUDIT_COLUMN_PREFIX]
+                        Set the audit column prefix, used in the destination
+                        schema. Default is 'rdl_'.
 ```
 
 _Notes:_

rdl/BatchDataLoader.py

@@ -3,7 +3,7 @@
 
 from io import StringIO
 from rdl.column_transformers.StringTransformers import ToUpper
-from rdl.shared import Constants
+from rdl.shared import Providers
 from rdl.shared.Utils import prevent_senstive_data_logging
 
 
@@ -97,7 +97,7 @@ def get_destination_column_name(self, source_column_name):
                 return column['destination']['name']
 
         # Audit columns - map them straight through
-        if source_column_name.startswith(Constants.AUDIT_COLUMN_PREFIX):
+        if source_column_name.startswith(Providers.AuditColumnsNames.audit_column_prefix):
             return source_column_name
 
         message = f"A source column with name '{source_column_name}' was not found in the column configuration"

rdl/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 rdl.shared import Constants
+from rdl.shared import Providers
 
 
 class DestinationTableManager(object):
@@ -39,13 +39,13 @@ def create_table(self, schema_name, table_name, columns_configuration, drop_firs
             table.append_column(self.create_column(column_configuration['destination']))
 
         table.append_column(
-            Column(Constants.AuditColumnNames.TIMESTAMP, DateTime(timezone=True), server_default=func.now()))
+            Column(Providers.AuditColumnsNames.TIMESTAMP, DateTime(timezone=True), server_default=func.now()))
 
         table.append_column(
-            Column(Constants.AuditColumnNames.IS_DELETED, Boolean, server_default='f', default=False))
+            Column(Providers.AuditColumnsNames.IS_DELETED, Boolean, server_default='f', default=False))
 
         table.append_column(
-            Column(Constants.AuditColumnNames.CHANGE_VERSION, BigInteger))
+            Column(Providers.AuditColumnsNames.CHANGE_VERSION, BigInteger))
 
         if drop_first:
             self.logger.debug(f"Dropping table {schema_name}.{table_name}")
@@ -103,9 +103,9 @@ def rename_table(self, schema_name, source_table_name, target_table_name):
     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}"
+        column_list = column_list + f",{Providers.AuditColumnsNames.TIMESTAMP}"
+        column_list = column_list + f",{Providers.AuditColumnsNames.IS_DELETED}"
+        column_list = column_list + f",{Providers.AuditColumnsNames.CHANGE_VERSION}"
 
         primary_key_column_array = [column_config['destination']['name'] for column_config in
                                     columns_config if 'primary_key' in column_config['destination'] and
@@ -121,9 +121,9 @@ def upsert_table(self, schema_name, source_table_name, target_table_name, column
         for column_config in columns_config:
             sql_builder.write("{0} = EXCLUDED.{0},\n".format(column_config['destination']['name']))
 
-        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))
+        sql_builder.write("{0} = EXCLUDED.{0},\n".format(Providers.AuditColumnsNames.TIMESTAMP))
+        sql_builder.write("{0} = EXCLUDED.{0},\n".format(Providers.AuditColumnsNames.IS_DELETED))
+        sql_builder.write("{0} = EXCLUDED.{0};\n".format(Providers.AuditColumnsNames.CHANGE_VERSION))
 
         upsert_sql = sql_builder.getvalue()
 

rdl/RelationalDataLoader.py

@@ -3,7 +3,7 @@
 from datetime import datetime
 from sqlalchemy import create_engine
 from rdl.DataLoadManager import DataLoadManager
-from rdl.shared import Constants
+from rdl.shared import Constants, Providers
 from rdl.data_load_tracking.DataLoadTrackerRepository import DataLoadTrackerRepository
 from rdl.data_sources.DataSourceFactory import DataSourceFactory
 from sqlalchemy.orm import sessionmaker
@@ -24,6 +24,7 @@ def main(self):
         self.args = self.get_arguments()
 
         self.configure_root_logger(self.args.log_level)
+        Providers.AuditColumnsNames.update_audit_column_prefix(self.args.audit_column_prefix)
 
         self.args.func()
 
@@ -130,6 +131,13 @@ def get_arguments(self):
             nargs='?',
             help=f'Set the logging output level. {_LOG_LEVEL_STRINGS}')
 
+        process_command_parser.add_argument(
+            '-p', '--audit-column-prefix',
+            default='rdl_',
+            type=str,
+            nargs='?',
+            help=f'Set the audit column prefix, used in the destination schema. Default is \'rdl_\'. ')
+
         audit_command_parser = subparsers.add_parser('audit',
                                                      help='provides list of processed models since a given timestamp')
         audit_command_parser.set_defaults(func=self.execute_audit_command)
@@ -162,6 +170,13 @@ def get_arguments(self):
             nargs='?',
             help=f'Set the logging output level. {_LOG_LEVEL_STRINGS}')
 
+        audit_command_parser.add_argument(
+            '-p', '--audit-column-prefix',
+            default='rdl_',
+            type=str,
+            nargs='?',
+            help=f'Set the audit column prefix, used in the destination schema. Default is \'rdl_\'. ')
+
         return parser.parse_args()
 
     def str2bool(self, v):

rdl/data_sources/MsSqlDataSource.py

@@ -12,7 +12,7 @@
 
 from rdl.ColumnTypeResolver import ColumnTypeResolver
 from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
-from rdl.shared import Constants
+from rdl.shared import Providers
 from rdl.shared.Utils import prevent_senstive_data_logging
 
 
@@ -97,9 +97,9 @@ def build_select_statement(self, table_config, columns, batch_config, batch_key_
         else:
             select_sql = f"SELECT TOP ({batch_config['size']}) {column_names}, " \
                 f"{MsSqlDataSource.CHANGE_TABLE_ALIAS}.SYS_CHANGE_VERSION" \
-                f" AS {Constants.AuditColumnNames.CHANGE_VERSION}, " \
+                f" AS {Providers.AuditColumnsNames.CHANGE_VERSION}, " \
                 f"CASE {MsSqlDataSource.CHANGE_TABLE_ALIAS}.SYS_CHANGE_OPERATION WHEN 'D' THEN 1 ELSE 0 " \
-                f"END AS {Constants.AuditColumnNames.IS_DELETED}"
+                f"END AS {Providers.AuditColumnsNames.IS_DELETED}"
             from_sql = f"FROM CHANGETABLE(CHANGES" \
                 f" {table_config['schema']}.{table_config['name']}," \
                 f" {change_tracking_info.last_sync_version})" \

rdl/shared/Constants.py

@@ -1,6 +1,5 @@
 APP_NAME = 'Relational Data Loader'
 DATA_PIPELINE_EXECUTION_SCHEMA_NAME = 'data_pipeline'
-AUDIT_COLUMN_PREFIX = f'{DATA_PIPELINE_EXECUTION_SCHEMA_NAME}_'
 
 
 class FullRefreshReason:
@@ -19,9 +18,3 @@ class ExecutionStatus:
     LOAD_COMPLETED_SUCCESSFULLY = 'Load Completed Successfully'
     SKIPPED_AS_ZERO_ROWS = 'Skipped - Zero Rows'
     COMPLETED_SUCCESSFULLY = 'Completed Successfully'
-
-
-class AuditColumnNames:
-    TIMESTAMP = f'{AUDIT_COLUMN_PREFIX}timestamp'
-    IS_DELETED = f'{AUDIT_COLUMN_PREFIX}is_deleted'
-    CHANGE_VERSION = f'{AUDIT_COLUMN_PREFIX}change_version'

rdl/shared/Providers.py

@@ -0,0 +1,25 @@
+class __AuditColumnsNames:
+    def __init__(self):
+        self.audit_column_prefix = 'rdl_'
+        self.__changed = False
+
+    def update_audit_column_prefix(self, audit_column_prefix):
+        if self.__changed:
+            raise RuntimeError("Audit Column Prefix has already been set")
+        self.audit_column_prefix = audit_column_prefix
+        self.__changed = True
+
+    @property
+    def TIMESTAMP(self):
+        return f'{self.audit_column_prefix}timestamp'
+
+    @property
+    def IS_DELETED(self):
+        return f'{self.audit_column_prefix}is_deleted'
+
+    @property
+    def CHANGE_VERSION(self):
+        return f'{self.audit_column_prefix}change_version'
+
+
+AuditColumnsNames = __AuditColumnsNames()