provide connectionfairy on initialize

This is so that dialect methods that are called within init
can assume the same argument structure as when they are called
in other places; we can nail down the type of object as well.

This change seems to mostly impact the isolation level routines
in the dialects, as these are called during initialize()
as well as on established connections.  these methods can now
assume a non-proxied DBAPI connection object in all cases,
as it is commonly required that attributes like ".autocommit"
are set on the object which don't work well in a proxied
situation.

Other changes:

* adds an interface for the "connectionfairy" concept
  called PoolProxiedConnection.
* Removes ``Connectable`` superclass of Connection.
  ``Connectable`` was originally meant to provide for the
  "method which accepts connection or engine" theme.  As this
  pattern  is greatly reduced in 2.0 and Engine no longer extends
  from it, the ``Connectable`` superclass doesnt serve any real
  purpose.

Leading from that, to set this in I also applied pep 484 annotations
to the Dialect base, and then in the interests of seeing some
of the typing information show up in my IDE did a little bit for Engine,
Connection and others.  I hope that it's feasible that we can
add annotations to specific classes and attributes ahead of when we
actually try to mass-populate the whole library.  This was
the original spirit of pep-484 that we can apply annotations
gradually.  I do of course want to try to do a mass-populate
although i think even in that case we will end up doing a lot
of manual work anyway (in particular for the changes here which
are distinct from what the stubs have).

Fixes: #7122
Change-Id: I5dd7fbff8a7ae520a81c165091af12a6a68826db

14 files changed, 95 insertions, 112 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 353c78c76..f0a7364a3 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -2766,25 +2766,25 @@ class MSDialect(default.DefaultDialect):
         ]
     )
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return list(self._isolation_lookup)
 
-    def set_isolation_level(self, connection, level):
-        cursor = connection.cursor()
-        cursor.execute("SET TRANSACTION ISOLATION LEVEL %s" % level)
+    def set_isolation_level(self, dbapi_connection, level):
+        cursor = dbapi_connection.cursor()
+        cursor.execute(f"SET TRANSACTION ISOLATION LEVEL {level}")
         cursor.close()
         if level == "SNAPSHOT":
-            connection.commit()
+            dbapi_connection.commit()
 
-    def get_isolation_level(self, connection):
+    def get_isolation_level(self, dbapi_connection):
         last_error = None
 
         views = ("sys.dm_exec_sessions", "sys.dm_pdw_nodes_exec_sessions")
         for view in views:
-            cursor = connection.cursor()
+            cursor = dbapi_connection.cursor()
             try:
                 cursor.execute(
-                    """
+                    f"""
                   SELECT CASE transaction_isolation_level
                     WHEN 0 THEN NULL
                     WHEN 1 THEN 'READ UNCOMMITTED'
@@ -2792,10 +2792,9 @@ class MSDialect(default.DefaultDialect):
                     WHEN 3 THEN 'REPEATABLE READ'
                     WHEN 4 THEN 'SERIALIZABLE'
                     WHEN 5 THEN 'SNAPSHOT' END AS TRANSACTION_ISOLATION_LEVEL
-                    FROM %s
+                    FROM {view}
                     where session_id = @@SPID
                   """
-                    % view
                 )
                 val = cursor.fetchone()[0]
             except self.dbapi.Error as err:
@@ -2811,12 +2810,12 @@ class MSDialect(default.DefaultDialect):
             # DefaultDialect, so the warning here is all that displays
             util.warn(
                 "Could not fetch transaction isolation level, "
-                "tried views: %s; final error was: %s" % (views, last_error)
+                f"tried views: {views}; final error was: {last_error}"
             )
             raise NotImplementedError(
                 "Can't fetch isolation level on this particular "
-                "SQL Server version. tried views: %s; final error was: %s"
-                % (views, last_error)
+                f"SQL Server version. tried views: {views}; final "
+                f"error was: {last_error}"
             )
 
     def initialize(self, connection):
diff --git a/lib/sqlalchemy/dialects/mssql/pymssql.py b/lib/sqlalchemy/dialects/mssql/pymssql.py
index 18bee1890..a9dc97d54 100644
--- a/lib/sqlalchemy/dialects/mssql/pymssql.py
+++ b/lib/sqlalchemy/dialects/mssql/pymssql.py
@@ -125,16 +125,18 @@ class MSDialect_pymssql(MSDialect):
         else:
             return False
 
-    def get_isolation_level_values(self, dbapi_conn):
-        return super().get_isolation_level_values(dbapi_conn) + ["AUTOCOMMIT"]
+    def get_isolation_level_values(self, dbapi_connection):
+        return super().get_isolation_level_values(dbapi_connection) + [
+            "AUTOCOMMIT"
+        ]
 
-    def set_isolation_level(self, connection, level):
+    def set_isolation_level(self, dbapi_connection, level):
         if level == "AUTOCOMMIT":
-            connection.autocommit(True)
+            dbapi_connection.autocommit(True)
         else:
-            connection.autocommit(False)
+            dbapi_connection.autocommit(False)
             super(MSDialect_pymssql, self).set_isolation_level(
-                connection, level
+                dbapi_connection, level
             )
 
 
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index f77d839f3..fef1ec81a 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2404,14 +2404,14 @@ class MySQLDialect(default.DefaultDialect):
             "REPEATABLE READ",
         )
 
-    def set_isolation_level(self, dbapi_conn, level):
-        cursor = dbapi_conn.cursor()
-        cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL %s" % level)
+    def set_isolation_level(self, dbapi_connection, level):
+        cursor = dbapi_connection.cursor()
+        cursor.execute(f"SET SESSION TRANSACTION ISOLATION LEVEL {level}")
         cursor.execute("COMMIT")
         cursor.close()
 
-    def get_isolation_level(self, connection):
-        cursor = connection.cursor()
+    def get_isolation_level(self, dbapi_connection):
+        cursor = dbapi_connection.cursor()
         if self._is_mysql and self.server_version_info >= (5, 7, 20):
             cursor.execute("SELECT @@transaction_isolation")
         else:
diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index 1e57c779d..40f1207d0 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -308,7 +308,7 @@ class MySQLDialect_mysqldb(MySQLDialect):
         else:
             return cset_name()
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return (
             "SERIALIZABLE",
             "READ UNCOMMITTED",
@@ -317,13 +317,13 @@ class MySQLDialect_mysqldb(MySQLDialect):
             "AUTOCOMMIT",
         )
 
-    def set_isolation_level(self, dbapi_conn, level):
+    def set_isolation_level(self, dbapi_connection, level):
         if level == "AUTOCOMMIT":
-            dbapi_conn.autocommit(True)
+            dbapi_connection.autocommit(True)
         else:
-            dbapi_conn.autocommit(False)
+            dbapi_connection.autocommit(False)
             super(MySQLDialect_mysqldb, self).set_isolation_level(
-                dbapi_conn, level
+                dbapi_connection, level
             )
 
 
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 7df2422b3..63131bf95 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1576,7 +1576,7 @@ class OracleDialect(default.DefaultDialect):
             # use the default
             return None
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return ["READ COMMITTED", "SERIALIZABLE"]
 
     def get_default_isolation_level(self, dbapi_conn):
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index a1d093886..eecf8567c 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -972,7 +972,7 @@ class OracleDialect_cx_oracle(OracleDialect):
         super(OracleDialect_cx_oracle, self).initialize(connection)
         self._detect_decimal_char(connection)
 
-    def get_isolation_level(self, connection):
+    def get_isolation_level(self, dbapi_connection):
         # sources:
 
         # general idea of transaction id, have to start one, etc.
@@ -984,7 +984,7 @@ class OracleDialect_cx_oracle(OracleDialect):
         # Oracle tuple comparison without using IN:
         # https://www.sql-workbench.eu/comparison/tuple_comparison.html
 
-        with connection.cursor() as cursor:
+        with dbapi_connection.cursor() as cursor:
             # this is the only way to ensure a transaction is started without
             # actually running DML.   There's no way to see the configured
             # isolation level without getting it from v$transaction which
@@ -1019,21 +1019,19 @@ class OracleDialect_cx_oracle(OracleDialect):
 
         return result
 
-    def get_isolation_level_values(self, dbapi_conn):
-        return super().get_isolation_level_values(dbapi_conn) + ["AUTOCOMMIT"]
+    def get_isolation_level_values(self, dbapi_connection):
+        return super().get_isolation_level_values(dbapi_connection) + [
+            "AUTOCOMMIT"
+        ]
 
-    def set_isolation_level(self, connection, level):
-        if hasattr(connection, "dbapi_connection"):
-            dbapi_connection = connection.dbapi_connection
-        else:
-            dbapi_connection = connection
+    def set_isolation_level(self, dbapi_connection, level):
         if level == "AUTOCOMMIT":
             dbapi_connection.autocommit = True
         else:
             dbapi_connection.autocommit = False
-            connection.rollback()
-            with connection.cursor() as cursor:
-                cursor.execute("ALTER SESSION SET ISOLATION_LEVEL=%s" % level)
+            dbapi_connection.rollback()
+            with dbapi_connection.cursor() as cursor:
+                cursor.execute(f"ALTER SESSION SET ISOLATION_LEVEL={level}")
 
     def _detect_decimal_char(self, connection):
         # we have the option to change this setting upon connect,
diff --git a/lib/sqlalchemy/dialects/postgresql/_psycopg_common.py b/lib/sqlalchemy/dialects/postgresql/_psycopg_common.py
index d82d5f009..a3a378947 100644
--- a/lib/sqlalchemy/dialects/postgresql/_psycopg_common.py
+++ b/lib/sqlalchemy/dialects/postgresql/_psycopg_common.py
@@ -150,7 +150,7 @@ class _PGDialect_common_psycopg(PGDialect):
             # requires that "dsn" be present as a blank string.
             return ([""], opts)
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return (
             "AUTOCOMMIT",
             "READ COMMITTED",
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index 1fdb46b6f..4951107bc 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -871,11 +871,11 @@ class PGDialect_asyncpg(PGDialect):
             "SERIALIZABLE": "serializable",
         }
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return list(self._isolation_lookup)
 
-    def set_isolation_level(self, connection, level):
-        connection.set_isolation_level(self._isolation_lookup[level])
+    def set_isolation_level(self, dbapi_connection, level):
+        dbapi_connection.set_isolation_level(self._isolation_lookup[level])
 
     def set_readonly(self, connection, value):
         connection.readonly = value
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index d1d881dc3..614c84b56 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3251,17 +3251,17 @@ class PGDialect(default.DefaultDialect):
             "REPEATABLE READ",
         )
 
-    def set_isolation_level(self, connection, level):
-        cursor = connection.cursor()
+    def set_isolation_level(self, dbapi_connection, level):
+        cursor = dbapi_connection.cursor()
         cursor.execute(
             "SET SESSION CHARACTERISTICS AS TRANSACTION "
-            "ISOLATION LEVEL %s" % level
+            f"ISOLATION LEVEL {level}"
         )
         cursor.execute("COMMIT")
         cursor.close()
 
-    def get_isolation_level(self, connection):
-        cursor = connection.cursor()
+    def get_isolation_level(self, dbapi_connection):
+        cursor = dbapi_connection.cursor()
         cursor.execute("show transaction isolation level")
         val = cursor.fetchone()[0]
         cursor.close()
diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index ede953195..1904a1ae1 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -446,7 +446,7 @@ class PGDialect_pg8000(PGDialect):
         # connection was closed normally
         return "connection is closed" in str(e)
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return (
             "AUTOCOMMIT",
             "READ COMMITTED",
@@ -455,21 +455,17 @@ class PGDialect_pg8000(PGDialect):
             "SERIALIZABLE",
         )
 
-    def set_isolation_level(self, connection, level):
+    def set_isolation_level(self, dbapi_connection, level):
         level = level.replace("_", " ")
 
-        # adjust for ConnectionFairy possibly being present
-        if hasattr(connection, "dbapi_connection"):
-            connection = connection.dbapi_connection
-
         if level == "AUTOCOMMIT":
-            connection.autocommit = True
+            dbapi_connection.autocommit = True
         else:
-            connection.autocommit = False
-            cursor = connection.cursor()
+            dbapi_connection.autocommit = False
+            cursor = dbapi_connection.cursor()
             cursor.execute(
                 "SET SESSION CHARACTERISTICS AS TRANSACTION "
-                "ISOLATION LEVEL %s" % level
+                f"ISOLATION LEVEL {level}"
             )
             cursor.execute("COMMIT")
             cursor.close()
@@ -516,13 +512,13 @@ class PGDialect_pg8000(PGDialect):
 
         return val == "on"
 
-    def set_client_encoding(self, connection, client_encoding):
-        # adjust for ConnectionFairy possibly being present
-        if hasattr(connection, "dbapi_connection"):
-            connection = connection.dbapi_connection
-
-        cursor = connection.cursor()
-        cursor.execute("SET CLIENT_ENCODING TO '" + client_encoding + "'")
+    def _set_client_encoding(self, dbapi_connection, client_encoding):
+        cursor = dbapi_connection.cursor()
+        cursor.execute(
+            f"""SET CLIENT_ENCODING TO '{
+            client_encoding.replace("'", "''")
+        }'"""
+        )
         cursor.execute("COMMIT")
         cursor.close()
 
@@ -556,7 +552,7 @@ class PGDialect_pg8000(PGDialect):
         if self.client_encoding is not None:
 
             def on_connect(conn):
-                self.set_client_encoding(conn, self.client_encoding)
+                self._set_client_encoding(conn, self.client_encoding)
 
             fns.append(on_connect)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg.py b/lib/sqlalchemy/dialects/postgresql/psycopg.py
index c2017c975..4219b4fff 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg.py
@@ -322,9 +322,6 @@ class PGDialect_psycopg(_PGDialect_common_psycopg):
         connection.isolation_level = isolation_level
 
     def get_isolation_level(self, dbapi_connection):
-        if hasattr(dbapi_connection, "dbapi_connection"):
-            dbapi_connection = dbapi_connection.dbapi_connection
-
         status_before = dbapi_connection.info.transaction_status
         value = super().get_isolation_level(dbapi_connection)
 
@@ -334,15 +331,14 @@ class PGDialect_psycopg(_PGDialect_common_psycopg):
             dbapi_connection.rollback()
         return value
 
-    def set_isolation_level(self, connection, level):
-        connection = getattr(connection, "dbapi_connection", connection)
+    def set_isolation_level(self, dbapi_connection, level):
         if level == "AUTOCOMMIT":
             self._do_isolation_level(
-                connection, autocommit=True, isolation_level=None
+                dbapi_connection, autocommit=True, isolation_level=None
             )
         else:
             self._do_isolation_level(
-                connection,
+                dbapi_connection,
                 autocommit=False,
                 isolation_level=self._isolation_lookup[level],
             )
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index 3d9f90a29..0bbad3257 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -597,7 +597,8 @@ class PGDialect_psycopg2(_PGDialect_common_psycopg):
         super(PGDialect_psycopg2, self).initialize(connection)
         self._has_native_hstore = (
             self.use_native_hstore
-            and self._hstore_oids(connection.connection) is not None
+            and self._hstore_oids(connection.connection.dbapi_connection)
+            is not None
         )
 
         # PGDialect.initialize() checks server version for <= 8.2 and sets
@@ -639,8 +640,8 @@ class PGDialect_psycopg2(_PGDialect_common_psycopg):
             "SERIALIZABLE": extensions.ISOLATION_LEVEL_SERIALIZABLE,
         }
 
-    def set_isolation_level(self, connection, level):
-        connection.set_isolation_level(self._isolation_lookup[level])
+    def set_isolation_level(self, dbapi_connection, level):
+        dbapi_connection.set_isolation_level(self._isolation_lookup[level])
 
     def set_readonly(self, connection, value):
         connection.readonly = value
@@ -660,47 +661,47 @@ class PGDialect_psycopg2(_PGDialect_common_psycopg):
         fns = []
         if self.client_encoding is not None:
 
-            def on_connect(conn):
-                conn.set_client_encoding(self.client_encoding)
+            def on_connect(dbapi_conn):
+                dbapi_conn.set_client_encoding(self.client_encoding)
 
             fns.append(on_connect)
 
         if self.dbapi and self.use_native_uuid:
 
-            def on_connect(conn):
-                extras.register_uuid(None, conn)
+            def on_connect(dbapi_conn):
+                extras.register_uuid(None, dbapi_conn)
 
             fns.append(on_connect)
 
         if self.dbapi and self.use_native_hstore:
 
-            def on_connect(conn):
-                hstore_oids = self._hstore_oids(conn)
+            def on_connect(dbapi_conn):
+                hstore_oids = self._hstore_oids(dbapi_conn)
                 if hstore_oids is not None:
                     oid, array_oid = hstore_oids
                     kw = {"oid": oid}
                     kw["array_oid"] = array_oid
-                    extras.register_hstore(conn, **kw)
+                    extras.register_hstore(dbapi_conn, **kw)
 
             fns.append(on_connect)
 
         if self.dbapi and self._json_deserializer:
 
-            def on_connect(conn):
+            def on_connect(dbapi_conn):
                 extras.register_default_json(
-                    conn, loads=self._json_deserializer
+                    dbapi_conn, loads=self._json_deserializer
                 )
                 extras.register_default_jsonb(
-                    conn, loads=self._json_deserializer
+                    dbapi_conn, loads=self._json_deserializer
                 )
 
             fns.append(on_connect)
 
         if fns:
 
-            def on_connect(conn):
+            def on_connect(dbapi_conn):
                 for fn in fns:
-                    fn(conn)
+                    fn(dbapi_conn)
 
             return on_connect
         else:
@@ -781,11 +782,10 @@ class PGDialect_psycopg2(_PGDialect_common_psycopg):
         )
 
     @util.memoized_instancemethod
-    def _hstore_oids(self, conn):
+    def _hstore_oids(self, dbapi_connection):
+
         extras = self._psycopg2_extras
-        if hasattr(conn, "dbapi_connection"):
-            conn = conn.dbapi_connection
-        oids = extras.HstoreAdapter.get_oids(conn)
+        oids = extras.HstoreAdapter.get_oids(dbapi_connection)
         if oids is not None and oids[0]:
             return oids[0:2]
         else:
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 3dea23e18..0c7f8d839 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1915,18 +1915,18 @@ class SQLiteDialect(default.DefaultDialect):
         {"READ UNCOMMITTED": 1, "SERIALIZABLE": 0}
     )
 
-    def get_isolation_level_values(self, dbapi_conn):
+    def get_isolation_level_values(self, dbapi_connection):
         return list(self._isolation_lookup)
 
-    def set_isolation_level(self, connection, level):
+    def set_isolation_level(self, dbapi_connection, level):
         isolation_level = self._isolation_lookup[level]
 
-        cursor = connection.cursor()
-        cursor.execute("PRAGMA read_uncommitted = %d" % isolation_level)
+        cursor = dbapi_connection.cursor()
+        cursor.execute(f"PRAGMA read_uncommitted = {isolation_level}")
         cursor.close()
 
-    def get_isolation_level(self, connection):
-        cursor = connection.cursor()
+    def get_isolation_level(self, dbapi_connection):
+        cursor = dbapi_connection.cursor()
         cursor.execute("PRAGMA read_uncommitted")
         res = cursor.fetchone()
         if res:
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
index 45a35be65..944d714a3 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
@@ -489,18 +489,14 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
         }
     )
 
-    def set_isolation_level(self, connection, level):
-        if hasattr(connection, "dbapi_connection"):
-            dbapi_connection = connection.dbapi_connection
-        else:
-            dbapi_connection = connection
+    def set_isolation_level(self, dbapi_connection, level):
 
         if level == "AUTOCOMMIT":
             dbapi_connection.isolation_level = None
         else:
             dbapi_connection.isolation_level = ""
             return super(SQLiteDialect_pysqlite, self).set_isolation_level(
-                connection, level
+                dbapi_connection, level
             )
 
     def on_connect(self):
@@ -509,11 +505,7 @@ class SQLiteDialect_pysqlite(SQLiteDialect):
                 return None
             return re.search(a, b) is not None
 
-        def set_regexp(connection):
-            if hasattr(connection, "dbapi_connection"):
-                dbapi_connection = connection.dbapi_connection
-            else:
-                dbapi_connection = connection
+        def set_regexp(dbapi_connection):
             dbapi_connection.create_function(
                 "regexp",
                 2,