@@ -28,15 +28,15 @@ class MsSqlDataSource(object):
2828 def __init__ (self , connection_string , logger = None ):
2929 self .logger = logger or logging .getLogger (__name__ )
3030 self .connection_string = connection_string
31- self .database_engine = create_engine (connection_string , creator = self .create_connection_with_failover )
31+ self .database_engine = create_engine (connection_string , creator = self .__create_connection_with_failover )
3232 self .column_type_resolver = ColumnTypeResolver ()
3333
3434 @staticmethod
3535 def can_handle_connection_string (connection_string ):
36- return MsSqlDataSource .connection_string_regex_match (connection_string ) is not None
36+ return MsSqlDataSource .__connection_string_regex_match (connection_string ) is not None
3737
3838 @staticmethod
39- def connection_string_regex_match (connection_string ):
39+ def __connection_string_regex_match (connection_string ):
4040 return re .match (MsSqlDataSource .MSSQL_STRING_REGEX , connection_string )
4141
4242 @staticmethod
@@ -52,8 +52,8 @@ def prefix_column(column_name, full_refresh, primary_key_column_names):
5252 else :
5353 return f"{ MsSqlDataSource .SOURCE_TABLE_ALIAS } .{ column_name } "
5454
55- def create_connection_with_failover (self ):
56- conn_string_data = MsSqlDataSource .connection_string_regex_match (self .connection_string )
55+ def __create_connection_with_failover (self ):
56+ conn_string_data = MsSqlDataSource .__connection_string_regex_match (self .connection_string )
5757 server = conn_string_data .group ('server' )
5858 failover = conn_string_data .group ('failover' )
5959 database = conn_string_data .group ('database' )
@@ -82,7 +82,7 @@ def create_connection_with_failover(self):
8282 return pyodbc .connect (dsn , server = failover )
8383 raise e
8484
85- def build_select_statement (self , table_config , columns , batch_config , batch_key_tracker , full_refresh ,
85+ def __build_select_statement (self , table_config , columns , batch_config , batch_key_tracker , full_refresh ,
8686 change_tracking_info ):
8787 column_array = list (
8888 map (lambda cfg : self .prefix_column (cfg ['source_name' ], full_refresh , table_config ['primary_keys' ]),
@@ -92,7 +92,7 @@ def build_select_statement(self, table_config, columns, batch_config, batch_key_
9292 if full_refresh :
9393 select_sql = f"SELECT TOP ({ batch_config ['size' ]} ) { column_names } "
9494 from_sql = f"FROM { table_config ['schema' ]} .{ table_config ['name' ]} AS { MsSqlDataSource .SOURCE_TABLE_ALIAS } "
95- where_sql = f"WHERE { self .build_where_clause (batch_key_tracker , MsSqlDataSource .SOURCE_TABLE_ALIAS )} "
95+ where_sql = f"WHERE { self .__build_where_clause (batch_key_tracker , MsSqlDataSource .SOURCE_TABLE_ALIAS )} "
9696 order_by_sql = "ORDER BY " + f", { MsSqlDataSource .SOURCE_TABLE_ALIAS } ." .join (table_config ['primary_keys' ])
9797 else :
9898 select_sql = f"SELECT TOP ({ batch_config ['size' ]} ) { column_names } , " \
@@ -105,29 +105,29 @@ def build_select_statement(self, table_config, columns, batch_config, batch_key_
105105 f" { change_tracking_info .last_sync_version } )" \
106106 f" AS { MsSqlDataSource .CHANGE_TABLE_ALIAS } " \
107107 f" LEFT JOIN { table_config ['schema' ]} .{ table_config ['name' ]} AS { MsSqlDataSource .SOURCE_TABLE_ALIAS } " \
108- f" ON { self .build_change_table_on_clause (batch_key_tracker )} "
109- where_sql = f"WHERE { self .build_where_clause (batch_key_tracker , MsSqlDataSource .CHANGE_TABLE_ALIAS )} "
108+ f" ON { self .__build_change_table_on_clause (batch_key_tracker )} "
109+ where_sql = f"WHERE { self .__build_where_clause (batch_key_tracker , MsSqlDataSource .CHANGE_TABLE_ALIAS )} "
110110 order_by_sql = "ORDER BY " + f", { MsSqlDataSource .CHANGE_TABLE_ALIAS } ." .join (table_config ['primary_keys' ])
111111
112112 return f"{ select_sql } \n { from_sql } \n { where_sql } \n { order_by_sql } ;"
113113
114114 # Returns an array of configured_columns containing only columns that this data source supports. Logs invalid ones.
115115 def assert_data_source_is_valid (self , table_config , configured_columns ):
116- columns_in_database = self .get_table_columns (table_config )
116+ columns_in_database = self .__get_table_columns (table_config )
117117
118118 for column in configured_columns :
119- self .assert_column_exists (column ['source_name' ],
120- columns_in_database ,
121- f"{ table_config ['schema' ]} .{ table_config ['name' ]} " )
119+ self .__assert_column_exists (column ['source_name' ],
120+ columns_in_database ,
121+ f"{ table_config ['schema' ]} .{ table_config ['name' ]} " )
122122
123- def assert_column_exists (self , column_name , columns_in_database , table_name ):
123+ def __assert_column_exists (self , column_name , columns_in_database , table_name ):
124124 if column_name in columns_in_database :
125125 return True
126126
127127 message = f'Column { column_name } does not exist in source table { table_name } '
128128 raise ValueError (message )
129129
130- def get_table_columns (self , table_config ):
130+ def __get_table_columns (self , table_config ):
131131 metadata = MetaData ()
132132 self .logger .debug (f"Reading definition for source table "
133133 f"{ table_config ['schema' ]} .{ table_config ['name' ]} " )
@@ -138,7 +138,7 @@ def get_table_columns(self, table_config):
138138 @prevent_senstive_data_logging
139139 def get_next_data_frame (self , table_config , columns , batch_config , batch_tracker , batch_key_tracker ,
140140 full_refresh , change_tracking_info ):
141- sql = self .build_select_statement (table_config , columns , batch_config , batch_key_tracker ,
141+ sql = self .__build_select_statement (table_config , columns , batch_config , batch_key_tracker ,
142142 full_refresh , change_tracking_info )
143143
144144 self .logger .debug (f"Starting read of SQL Statement: \n { sql } " )
@@ -226,7 +226,7 @@ def get_change_tracking_info(self, table_config, last_known_sync_version):
226226 row ["force_full_load" ], row ["data_changed_since_last_sync" ])
227227
228228 @staticmethod
229- def build_where_clause (batch_key_tracker , table_alias ):
229+ def __build_where_clause (batch_key_tracker , table_alias ):
230230 has_value = False
231231
232232 try :
@@ -244,7 +244,7 @@ def build_where_clause(batch_key_tracker, table_alias):
244244 sql_builder .close ()
245245
246246 @staticmethod
247- def build_change_table_on_clause (batch_key_tracker ):
247+ def __build_change_table_on_clause (batch_key_tracker ):
248248 has_value = False
249249
250250 try :
0 commit comments