fix: blue green deployment test

Author: karenc-bq

aws_advanced_python_wrapper/mysql_driver_dialect.py

@@ -34,7 +34,7 @@
 
     CMYSQL_ENABLED = True
 
-except ImportError as exc:
+except ImportError:
     # Do nothing
     pass
 
@@ -175,8 +175,7 @@ def get_connection_from_obj(self, obj: object) -> Any:
         return None
 
     def transfer_session_state(self, from_conn: Connection, to_conn: Connection):
-        if (isinstance(from_conn, CMySQLConnection) or isinstance(from_conn, MySQLConnection)) and (
-                isinstance(to_conn, CMySQLConnection) or isinstance(to_conn, MySQLConnection)):
+        if MySQLDriverDialect._is_mysql_connection(from_conn) and MySQLDriverDialect._is_mysql_connection(to_conn):
             to_conn.autocommit = from_conn.autocommit
 
     def ping(self, conn: Connection) -> bool:

tests/integration/container/test_blue_green_deployment.py

@@ -30,7 +30,6 @@
 
 import mysql.connector
 import psycopg
-from mysql.connector import CMySQLConnection, MySQLConnection
 
 from aws_advanced_python_wrapper.mysql_driver_dialect import MySQLDriverDialect
 from aws_advanced_python_wrapper.pg_driver_dialect import PgDriverDialect
@@ -458,7 +457,7 @@ def close_connection(self, conn: Optional[Connection]):
     def is_closed(self, conn: Connection) -> bool:
         if isinstance(conn, psycopg.Connection):
             return self.pg_dialect.is_closed(conn)
-        elif isinstance(conn, CMySQLConnection) or isinstance(conn, MySQLConnection):
+        elif MySQLDriverDialect._is_mysql_connection(conn):
             return self.mysql_dialect.is_closed(conn)
         elif isinstance(conn, AwsWrapperConnection):
             return conn.is_closed