Genericize setinputsizes and support pyodbc

Reworked the "setinputsizes()" set of dialect hooks to be correctly
extensible for any arbirary DBAPI, by allowing dialects individual hooks
that may invoke cursor.setinputsizes() in the appropriate style for that
DBAPI.   In particular this is intended to support pyodbc's style of usage
which is fundamentally different from that of cx_Oracle.  Added support
for pyodbc.

Fixes: #5649
Change-Id: I9f1794f8368bf3663a286932cfe3992dae244a10

3 files changed, 66 insertions, 8 deletions

diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index d1b69100f..7bde19090 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -687,15 +687,12 @@ class OracleExecutionContext_cx_oracle(OracleExecutionContext):
         if self.compiled._quoted_bind_names:
             self._setup_quoted_bind_names()
 
-        self.set_input_sizes(
-            self.compiled._quoted_bind_names,
-            include_types=self.dialect._include_setinputsizes,
-        )
-
         self._generate_out_parameter_vars()
 
         self._generate_cursor_outputtype_handler()
 
+        self.include_set_input_sizes = self.dialect._include_setinputsizes
+
     def post_exec(self):
         if self.compiled and self.out_parameters and self.compiled.returning:
             # create a fake cursor result from the out parameters. unlike
@@ -746,6 +743,8 @@ class OracleDialect_cx_oracle(OracleDialect):
     supports_unicode_statements = True
     supports_unicode_binds = True
 
+    use_setinputsizes = True
+
     driver = "cx_oracle"
 
     colspecs = {
@@ -1172,6 +1171,35 @@ class OracleDialect_cx_oracle(OracleDialect):
             if oci_prepared:
                 self.do_commit(connection.connection)
 
+    def do_set_input_sizes(self, cursor, list_of_tuples, context):
+        if self.positional:
+            # not usually used, here to support if someone is modifying
+            # the dialect to use positional style
+            cursor.setinputsizes(
+                *[dbtype for key, dbtype, sqltype in list_of_tuples]
+            )
+        else:
+            collection = (
+                (key, dbtype)
+                for key, dbtype, sqltype in list_of_tuples
+                if dbtype
+            )
+            if context and context.compiled:
+                quoted_bind_names = context.compiled._quoted_bind_names
+                collection = (
+                    (quoted_bind_names.get(key, key), dbtype)
+                    for key, dbtype in collection
+                )
+
+            if not self.supports_unicode_binds:
+                # oracle 8 only
+                collection = (
+                    (self.dialect._encoder(key)[0], dbtype)
+                    for key, dbtype in collection
+                )
+
+            cursor.setinputsizes(**{key: dbtype for key, dbtype in collection})
+
     def do_recover_twophase(self, connection):
         connection.info.pop("cx_oracle_prepared", None)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index a4937d0d2..7d679731b 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -245,7 +245,7 @@ class PGExecutionContext_asyncpg(PGExecutionContext):
         # we have to exclude ENUM because "enum" not really a "type"
         # we can cast to, it has to be the name of the type itself.
         # for now we just omit it from casting
-        self.set_input_sizes(exclude_types={AsyncAdapt_asyncpg_dbapi.ENUM})
+        self.exclude_set_input_sizes = {AsyncAdapt_asyncpg_dbapi.ENUM}
 
     def create_server_side_cursor(self):
         return self._dbapi_connection.cursor(server_side=True)
@@ -687,6 +687,8 @@ class PGDialect_asyncpg(PGDialect):
     statement_compiler = PGCompiler_asyncpg
     preparer = PGIdentifierPreparer_asyncpg
 
+    use_setinputsizes = True
+
     use_native_uuid = True
 
     colspecs = util.update_copy(
@@ -787,6 +789,20 @@ class PGDialect_asyncpg(PGDialect):
                 e, self.dbapi.InterfaceError
             ) and "connection is closed" in str(e)
 
+    def do_set_input_sizes(self, cursor, list_of_tuples, context):
+        if self.positional:
+            cursor.setinputsizes(
+                *[dbtype for key, dbtype, sqltype in list_of_tuples]
+            )
+        else:
+            cursor.setinputsizes(
+                **{
+                    key: dbtype
+                    for key, dbtype, sqltype in list_of_tuples
+                    if dbtype
+                }
+            )
+
     def on_connect(self):
         super_connect = super(PGDialect_asyncpg, self).on_connect()
 
diff --git a/lib/sqlalchemy/dialects/postgresql/pg8000.py b/lib/sqlalchemy/dialects/postgresql/pg8000.py
index b2faa4243..439249157 100644
--- a/lib/sqlalchemy/dialects/postgresql/pg8000.py
+++ b/lib/sqlalchemy/dialects/postgresql/pg8000.py
@@ -234,8 +234,6 @@ class PGExecutionContext_pg8000(PGExecutionContext):
         if not self.compiled:
             return
 
-        self.set_input_sizes()
-
 
 class PGCompiler_pg8000(PGCompiler):
     def visit_mod_binary(self, binary, operator, **kw):
@@ -265,6 +263,8 @@ class PGDialect_pg8000(PGDialect):
     statement_compiler = PGCompiler_pg8000
     preparer = PGIdentifierPreparer_pg8000
 
+    use_setinputsizes = True
+
     # reversed as of pg8000 1.16.6.  1.16.5 and lower
     # are no longer compatible
     description_encoding = None
@@ -407,6 +407,20 @@ class PGDialect_pg8000(PGDialect):
         cursor.execute("COMMIT")
         cursor.close()
 
+    def do_set_input_sizes(self, cursor, list_of_tuples, context):
+        if self.positional:
+            cursor.setinputsizes(
+                *[dbtype for key, dbtype, sqltype in list_of_tuples]
+            )
+        else:
+            cursor.setinputsizes(
+                **{
+                    key: dbtype
+                    for key, dbtype, sqltype in list_of_tuples
+                    if dbtype
+                }
+            )
+
     def do_begin_twophase(self, connection, xid):
         connection.connection.tpc_begin((0, xid, ""))