[OSC-1257] add failover server compatibility (#25)

* add failover server compatibility

* better logging

* make username/password check stricter

* only log things that exist

* use trusted connection

* quote string in integration tests

* stricter regex

* simplify logic, complify regex

* tidy regex and sort imports

* avoid divide by zero error

* move function down

Author: seanbudd

appveyor.yml

@@ -66,6 +66,7 @@ test_script:
 
   - test_full_refresh_from_mssql.cmd
   - test_audit.cmd
+  - test_mssql_failover_server.cmd
 
 on_finish:
   #Enable this line to make the build pause after completion for RDP troubleshooting.

rdl/data_load_tracking/DataLoadTracker.py

@@ -47,7 +47,7 @@ def completed_successfully(self):
         for batch in self.batches:
             self.total_row_count += batch.row_count
 
-        self.rows_per_second = self.total_row_count / self.total_execution_time.total_seconds()
+        self.rows_per_second = self.total_row_count / (self.total_execution_time.total_seconds() + 1e-10)
 
     def get_statistics(self):
         load_type = 'FULL' if self.is_full_refresh else f"INCREMENTAL from " \

rdl/data_sources/MsSqlDataSource.py

@@ -1,28 +1,42 @@
 import io
 import logging
 import pandas
+import pyodbc
+import re
+
+import sqlalchemy.exc
 from sqlalchemy import create_engine
 from sqlalchemy import MetaData
 from sqlalchemy.schema import Table
+from sqlalchemy.sql import text
+
 from rdl.ColumnTypeResolver import ColumnTypeResolver
 from rdl.data_sources.ChangeTrackingInfo import ChangeTrackingInfo
-from sqlalchemy.sql import text
 from rdl.shared import Constants
 
 
 class MsSqlDataSource(object):
     SOURCE_TABLE_ALIAS = 'src'
     CHANGE_TABLE_ALIAS = 'chg'
+    MSSQL_STRING_REGEX = r"mssql\+pyodbc://" \
+        r"(?:(?P<username>[^@/?&:]+)?:(?P<password>[^@/?&:]+)?@)?" \
+        r"(?P<server>[^@/?&:]*)/(?P<database>[^@/?&:]*)" \
+        r"\?driver=(?P<driver>[^@/?&:]*)" \
+        r"(?:&failover=(?P<failover>[^@/?&:]*))?"
 
     def __init__(self, connection_string, logger=None):
         self.logger = logger or logging.getLogger(__name__)
         self.connection_string = connection_string
-        self.database_engine = create_engine(connection_string)
+        self.database_engine = create_engine(connection_string, creator=self.create_connection_with_failover)
         self.column_type_resolver = ColumnTypeResolver()
 
     @staticmethod
     def can_handle_connection_string(connection_string):
-        return connection_string.startswith(MsSqlDataSource.connection_string_prefix())
+        return MsSqlDataSource.connection_string_regex_match(connection_string) is not None
+
+    @staticmethod
+    def connection_string_regex_match(connection_string):
+        return re.match(MsSqlDataSource.MSSQL_STRING_REGEX, connection_string)
 
     @staticmethod
     def connection_string_prefix():
@@ -37,6 +51,36 @@ def prefix_column(column_name, full_refresh, primary_key_column_names):
         else:
             return f"{MsSqlDataSource.SOURCE_TABLE_ALIAS}.{column_name}"
 
+    def create_connection_with_failover(self):
+        conn_string_data = MsSqlDataSource.connection_string_regex_match(self.connection_string)
+        server = conn_string_data.group('server')
+        failover = conn_string_data.group('failover')
+        database = conn_string_data.group('database')
+        driver = "{"+conn_string_data.group('driver').replace('+', ' ')+"}"
+        dsn = f'DRIVER={driver};DATABASE={database};'
+
+        username = conn_string_data.group('username')
+        password = conn_string_data.group('password')
+
+        login_cred = "Trusted_Connection=yes;"
+        if username is not None and password is not None:
+            login_cred = f'UID={username};PWD={password};'
+
+        dsn += login_cred
+        self.logger.info(
+            'Parsed Connection Details: ' +
+            f'''FAILOVER={failover}
+            SERVER={server}
+            DRIVER={driver}
+            DATABASE={database}''')
+        try:
+            return pyodbc.connect(dsn, server=server)
+        except (sqlalchemy.exc.OperationalError, pyodbc.OperationalError) as e:
+            if e.args[0] == "08001" and failover is not None:
+                self.logger.warning(f'Using Failover Server: {failover}')
+                return pyodbc.connect(dsn, server=failover)
+            raise e
+
     def build_select_statement(self, table_config, columns, batch_config, batch_key_tracker, full_refresh,
                                change_tracking_info):
         column_array = list(

test_mssql_failover_server.cmd

@@ -0,0 +1,9 @@
+@echo off
+py -m rdl process "mssql+pyodbc://(local)\SQL2016/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0" postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG
+if %errorlevel% neq 0 exit /b %errorlevel%
+
+py -m rdl process "mssql+pyodbc://fake_server\SQL2016/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0&failover=(local)" postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG
+if %errorlevel% neq 0 exit /b %errorlevel%
+
+py -m rdl process "mssql+pyodbc://(local)\SQL2016/RelationalDataLoaderIntegrationTestSource?driver=SQL+Server+Native+Client+11.0&failover=fake_server" postgresql+psycopg2://postgres:there_is_no_password_due_to_pg_trust@localhost/relational_data_loader_integration_tests ./integration_tests/mssql_source/config/ --log-level DEBUG
+if %errorlevel% neq 0 exit /b %errorlevel%