Code review fixups

Author: dames

integration_tests/csv_source/assertions/column_test_full_refresh_assertions.sql

@@ -4,7 +4,7 @@ DROP TABLE IF EXISTS results;
 
 CREATE TEMPORARY TABLE results AS
   WITH expected(id, int_column_1 , date_column_1, decimal_column_1, date_time_column_1, string_column_1) AS (
-    SELECT 1, 111.0, '1976-12-01'::date, 12.1212, '1976-12-01 01:00:00.000000'::timestamp, 'A Basic String'
+    SELECT 1, 111.0, '1976-12-01'::DATE, 12.1212, '1976-12-01 01:00:00.000000'::TIMESTAMP, 'A Basic String'
     UNION ALL
     SELECT 2, NULL, NULL, NULL, NULL, NULL
     UNION ALL

integration_tests/csv_source/assertions/column_test_incremental_refresh_assertions.sql

@@ -3,7 +3,7 @@ DROP TABLE IF EXISTS results;
 
 CREATE TEMPORARY TABLE results AS
   WITH expected(id, int_column_1 , date_column_1, decimal_column_1, date_time_column_1, string_column_1) AS (
-    SELECT 1, 111.0, '1976-12-01'::date, 12.1212, '1976-12-01 01:00:00.000000'::timestamp, 'A Basic String'
+    SELECT 1, 111.0, '1976-12-01'::DATE, 12.1212, '1976-12-01 01:00:00.000000'::TIMESTAMP, 'A Basic String'
     UNION ALL
     SELECT 2, NULL, NULL, NULL, NULL, NULL
     UNION ALL
@@ -13,9 +13,9 @@ CREATE TEMPORARY TABLE results AS
     UNION ALL
     SELECT 5, NULL, NULL, NULL, NULL, 'This row WAS updated in the incremental review test'
     UNION ALL
-    SELECT 6, 111.0, '1976-12-01'::date, 12.1212, '1976-12-01 01:00:00.000000'::timestamp, 'A Basic String'
+    SELECT 6, 111.0, '1976-12-01'::DATE, 12.1212, '1976-12-01 01:00:00.000000'::TIMESTAMP, 'A Basic String'
     UNION ALL
-    SELECT 7, 111.0, '1976-12-01'::date, 12.1212, '1976-12-01 01:00:00.000000'::timestamp, 'Another Basic String'
+    SELECT 7, 111.0, '1976-12-01'::DATE, 12.1212, '1976-12-01 01:00:00.000000'::TIMESTAMP, 'Another Basic String'
   ),
 
   actual AS (

integration_tests/mssql_source/source_database_setup/create_large_table.sql

@@ -1,17 +1,15 @@
 IF object_id('LargeTable') IS NULL
-	CREATE TABLE LargeTable
-	(	Id	int PRIMARY KEY,
-		StringColumn1 varchar(100),
-		DateColumn1	datetime,
-		DateColumn2	date,
-		IntColumn1	int,
-		StringColumn2 nvarchar(100),
-		GuidColumn uniqueidentifier)
+	CREATE TABLE LargeTable (
+      Id	int PRIMARY KEY,
+      StringColumn1 varchar(100),
+      DateColumn1	datetime,
+      DateColumn2	date,
+      IntColumn1 int,
+      StringColumn2 nvarchar(100),
+      GuidColumn uniqueidentifier)
 ELSE
 	TRUNCATE TABLE LargeTable
 
-
-
 ;WITH Numbers (n) AS
 (
     SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL))
@@ -25,22 +23,22 @@ ELSE
 
 INSERT LargeTable
 (
-	Id,
-	StringColumn1 ,
-	DateColumn1,
-	DateColumn2,
-	IntColumn1,
-	StringColumn2 ,
-	GuidColumn
+        Id,
+        StringColumn1,
+        DateColumn1,
+        DateColumn2,
+        IntColumn1,
+        StringColumn2,
+        GuidColumn
 )
 SELECT	n,
-		CASE WHEN n % 3 = 0 THEN NULL ELSE 'Row Number ' + CAST(n as varchar) END,
-		CASE WHEN n % 5 = 0 THEN NULL ELSE DateAdd(hour, -n, '2000-01-1') END,
-		CASE WHEN n % 7 = 0 THEN NULL ELSE DateAdd(hour, n, '2000-01-1') END,
-		CASE WHEN n % 9 = 0 THEN NULL ELSE n * 1000 END,
-		CASE WHEN n % 11 = 0 THEN NULL ELSE N'काचं शक्नोम्यत्तुम् । नोपहिनस्ति माम् ॥' END,
-		CASE WHEN n % 13 = 0 THEN NULL ELSE newid() END
-FROM  Numbers
+        CASE WHEN n % 3 = 0 THEN NULL ELSE 'Row Number ' + CAST(n as varchar) END,
+        CASE WHEN n % 5 = 0 THEN NULL ELSE DateAdd(hour, -n, '2000-01-1') END,
+        CASE WHEN n % 7 = 0 THEN NULL ELSE DateAdd(hour, n, '2000-01-1') END,
+        CASE WHEN n % 9 = 0 THEN NULL ELSE n * 1000 END,
+        CASE WHEN n % 11 = 0 THEN NULL ELSE N'काचं शक्नोम्यत्तुम् । नोपहिनस्ति माम् ॥' END,
+        CASE WHEN n % 13 = 0 THEN NULL ELSE newid() END
+FROM    Numbers
 
 
 

modules/BatchDataLoader.py

@@ -1,7 +1,6 @@
 import logging
-from io import StringIO
 import importlib
-
+from io import StringIO
 from modules.column_transformers.StringTransformers import ToUpper
 
 
@@ -31,7 +30,6 @@ def load_batch(self, previous_batch_key):
             return -1
 
         data_frame = self.attach_column_transformers(data_frame)
-
         self.write_data_frame_to_table(data_frame)
         batch_tracker.load_completed_successfully()
 

modules/DataLoadManager.py

@@ -1,9 +1,10 @@
+import os
+import json
 import logging
 from modules.BatchDataLoader import BatchDataLoader
 from modules.DestinationTableManager import DestinationTableManager
 from modules.DataLoadTracker import DataLoadTracker
-import os
-import json
+
 
 
 class DataLoadManager(object):
@@ -48,7 +49,7 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
 
         destination_table_manager.create_schema(pipeline_configuration['target_schema'])
 
-        self.logger.info("Recreating the staging table {0}.{1}".format(pipeline_configuration['target_schema'], pipeline_configuration['stage_table']))
+        self.logger.debug("Recreating the staging table {0}.{1}".format(pipeline_configuration['target_schema'], pipeline_configuration['stage_table']))
         destination_table_manager.create_table(pipeline_configuration['target_schema'],
                                                pipeline_configuration['stage_table'],
                                                columns, drop_first=True)
@@ -68,11 +69,9 @@ def start_single_import(self, target_engine, configuration_name, requested_full_
         while previous_unique_column_value > -1:
             previous_unique_column_value = batch_data_loader.load_batch(previous_unique_column_value)
 
-        self.logger.info("ImportBatch Completed")
-
         if full_refresh:
             # Rename the stage table to the load table.
-            self.logger.info("Full-load is set. Renaming the stage table to the load table.")
+            self.logger.debug("Full-load is set. Renaming the stage table to the load table.")
             destination_table_manager.rename_table(pipeline_configuration['target_schema'],
                                                    pipeline_configuration['stage_table'],
                                                    pipeline_configuration['load_table'])

modules/DestinationTableManager.py

@@ -1,11 +1,10 @@
+import io
+import os
+import importlib
 import logging
 from sqlalchemy import MetaData, DateTime
 from sqlalchemy.schema import Column, Table
-import importlib
 from sqlalchemy.sql import func
-import io
-import os
-
 
 class DestinationTableManager(object):
     TIMESTAMP_COLUMN_NAME = "data_pipeline_timestamp"

modules/RelationalDataLoader.py

@@ -1,10 +1,9 @@
 import logging
+import argparse
+from sqlalchemy import create_engine
 from modules.DataLoadManager import DataLoadManager
 from modules.data_sources.DataSourceFactory import DataSourceFactory
 
-from sqlalchemy import create_engine
-import argparse
-
 _LOG_LEVEL_STRINGS = ['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG']