[N/A] Reformat files with Black

Author: waltaro

rdl/BatchDataLoader.py

@@ -8,8 +8,20 @@
 
 
 class BatchDataLoader(object):
-    def __init__(self, source_db, source_table_config, target_schema, target_table, columns, data_load_tracker,
-                 batch_config, target_db, full_refresh, change_tracking_info, logger=None):
+    def __init__(
+        self,
+        source_db,
+        source_table_config,
+        target_schema,
+        target_table,
+        columns,
+        data_load_tracker,
+        batch_config,
+        target_db,
+        full_refresh,
+        change_tracking_info,
+        logger=None,
+    ):
         self.logger = logger or logging.getLogger(__name__)
         self.source_table_config = source_table_config
         self.columns = columns
@@ -26,14 +38,22 @@ def __init__(self, source_db, source_table_config, target_schema, target_table,
     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"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_table_data_frame(self.source_table_config, self.columns,
-                                                         self.batch_config, batch_tracker, batch_key_tracker,
-                                                         self.full_refresh, self.change_tracking_info)
+        self.logger.debug(
+            f"ImportBatch Starting from previous_batch_key: '{batch_key_tracker.bookmarks}'. "
+            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_table_data_frame(
+            self.source_table_config,
+            self.columns,
+            self.batch_config,
+            batch_tracker,
+            batch_key_tracker,
+            self.full_refresh,
+            self.change_tracking_info,
+        )
 
         if data_frame is None or len(data_frame) == 0:
             self.logger.debug("There are no more rows to import.")
@@ -46,21 +66,33 @@ def load_batch(self, batch_key_tracker):
         batch_tracker.load_completed_successfully()
 
         for primary_key in batch_key_tracker.primary_keys:
-            batch_key_tracker.set_bookmark(primary_key, int(data_frame.iloc[-1][primary_key]))
+            batch_key_tracker.set_bookmark(
+                primary_key, int(data_frame.iloc[-1][primary_key])
+            )
 
-        self.logger.info(f"Batch keys '{batch_key_tracker.bookmarks}' completed. {batch_tracker.get_statistics()}")
+        self.logger.info(
+            f"Batch keys '{batch_key_tracker.bookmarks}' completed. {batch_tracker.get_statistics()}"
+        )
 
     @prevent_senstive_data_logging
     def write_data_frame_to_table(self, data_frame):
-        qualified_target_table = f'{self.target_schema}.{self.target_table}'
+        qualified_target_table = f"{self.target_schema}.{self.target_table}"
         self.logger.debug(f"Starting write to table '{qualified_target_table}'")
         data = StringIO()
         # quoting: Due to \r existing in strings in MSSQL we must quote anything that's non numeric just to be safe
         # line_terminator: ensure \n is used even on windows machines as prod runs on *nix with \n
         # na_rep: Due to us quoting everything non-numeric, our null's must be represented by something special, as the
         # default null representation (nothing), once quoted, is equivalent to an empty string
-        data_frame.to_csv(data, header=False, index=False, na_rep='\\N', float_format='%.16g',
-                          quotechar='"', quoting=csv.QUOTE_NONNUMERIC, line_terminator='\n')
+        data_frame.to_csv(
+            data,
+            header=False,
+            index=False,
+            na_rep="\\N",
+            float_format="%.16g",
+            quotechar='"',
+            quoting=csv.QUOTE_NONNUMERIC,
+            line_terminator="\n",
+        )
         # Float_format is used to truncate any insignificant digits. Unfortunately it gives us an artificial limitation
 
         data.seek(0)
@@ -69,19 +101,27 @@ def write_data_frame_to_table(self, data_frame):
 
         # log CSV on debug
         if self.logger.getEffectiveLevel() == logging.DEBUG:
-            with open(f'{qualified_target_table}.csv', 'w', encoding='utf-8') as f:
+            with open(f"{qualified_target_table}.csv", "w", encoding="utf-8") as f:
                 f.write(data.getvalue())
 
         column_array = list(
-            map(lambda source_colum_name: self.get_destination_column_name(source_colum_name), data_frame.columns))
-        column_list = ','.join(map(str, column_array))
+            map(
+                lambda source_colum_name: self.get_destination_column_name(
+                    source_colum_name
+                ),
+                data_frame.columns,
+            )
+        )
+        column_list = ",".join(map(str, column_array))
 
         # FORCE_NULL: ensure quoted fields are checked for NULLs as by default they are assumed to be non-null
         # specify null as \N so that psql doesn't assume empty strings are nulls
-        sql = f"COPY {qualified_target_table}({column_list}) FROM STDIN "\
-            f"with (format csv, "\
-            f"null '\\N', "\
+        sql = (
+            f"COPY {qualified_target_table}({column_list}) FROM STDIN "
+            f"with (format csv, "
+            f"null '\\N', "
             f"FORCE_NULL ({column_list}))"
+        )
         self.logger.debug(f"Writing to table using command '{sql}'")
 
         curs.copy_expert(sql=sql, file=data)
@@ -93,11 +133,13 @@ def write_data_frame_to_table(self, data_frame):
 
     def get_destination_column_name(self, source_column_name):
         for column in self.columns:
-            if column['source_name'] == source_column_name:
-                return column['destination']['name']
+            if column["source_name"] == source_column_name:
+                return column["destination"]["name"]
 
         # Audit columns - map them straight through
-        if source_column_name.startswith(Providers.AuditColumnsNames.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"
@@ -106,9 +148,11 @@ def get_destination_column_name(self, source_column_name):
     def attach_column_transformers(self, data_frame):
         self.logger.debug("Attaching column transformers")
         for column in self.columns:
-            if 'column_transformer' in column:
+            if "column_transformer" in column:
                 # transformer = Utils.create_type_instance(column['column_transformer'])
                 transformer = ToUpper.execute
-                data_frame[column['source_name']] = data_frame[column['source_name']].map(transformer)
+                data_frame[column["source_name"]] = data_frame[
+                    column["source_name"]
+                ].map(transformer)
                 # print (data_frame)
         return data_frame

rdl/BatchKeyTracker.py

@@ -1,6 +1,4 @@
-
 class BatchKeyTracker(object):
-
     def __init__(self, primary_keys):
         self.primary_keys = primary_keys
         self.has_more_data = True

rdl/ColumnTypeResolver.py

@@ -5,37 +5,43 @@
 
 
 class ColumnTypeResolver(object):
-    PANDAS_TYPE_MAP = {'string': str,
-                       'datetime': str,
-                       'json': str,
-                       'numeric': float,
-                       'guid': str,
-                       'bigint': int,
-                       'boolean': bool}
+    PANDAS_TYPE_MAP = {
+        "string": str,
+        "datetime": str,
+        "json": str,
+        "numeric": float,
+        "guid": str,
+        "bigint": int,
+        "boolean": bool,
+    }
 
-    POSTGRES_TYPE_MAP = {'string': citext.CIText,
-                         'datetime': DateTime,
-                         'json': JSONB,
-                         'numeric': Numeric,
-                         'guid': UUID,
-                         'int': Integer,
-                         'bigint': BigInteger,
-                         'boolean': Boolean}
+    POSTGRES_TYPE_MAP = {
+        "string": citext.CIText,
+        "datetime": DateTime,
+        "json": JSONB,
+        "numeric": Numeric,
+        "guid": UUID,
+        "int": Integer,
+        "bigint": BigInteger,
+        "boolean": Boolean,
+    }
 
     def resolve_postgres_type(self, column):
-        return self.POSTGRES_TYPE_MAP[column['type']]
+        return self.POSTGRES_TYPE_MAP[column["type"]]
 
     def resolve_pandas_type(self, column):
-        if column['type'] == 'int':
-            if column['nullable']:
+        if column["type"] == "int":
+            if column["nullable"]:
                 return object
             else:
                 return int
         else:
-            return self.PANDAS_TYPE_MAP[column['type']]
+            return self.PANDAS_TYPE_MAP[column["type"]]
 
     def create_column_type_dictionary(self, columns):
         types = {}
         for column in columns:
-            types[column['source_name']] = self.resolve_pandas_type(column['destination'])
+            types[column["source_name"]] = self.resolve_pandas_type(
+                column["destination"]
+            )
         return types