def is_dialect()

in aws_advanced_python_wrapper/database_dialect.py [0:0]

• 23 lines of code
• 9 McCabe index (conditional complexity)

    def is_dialect(self, conn: Connection, driver_dialect: DriverDialect) -> bool:
        if not super().is_dialect(conn, driver_dialect):
            return False

        has_extensions: bool = False
        has_topology: bool = False

        initial_transaction_status: bool = driver_dialect.is_in_transaction(conn)
        try:
            with closing(conn.cursor()) as cursor:
                cursor.execute(self._EXTENSIONS_QUERY)
                row = cursor.fetchone()
                if row and bool(row[0]):
                    logger.debug("AuroraPgDialect.HasExtensionsTrue")
                    has_extensions = True

            with closing(conn.cursor()) as cursor:
                cursor.execute(self._HAS_TOPOLOGY_QUERY)
                if cursor.fetchone() is not None:
                    logger.debug("AuroraPgDialect.HasTopologyTrue")
                    has_topology = True

            return has_extensions and has_topology
        except Exception:
            if not initial_transaction_status and driver_dialect.is_in_transaction(conn):
                conn.rollback()

        return False