Support TypeDecorator.get_dbapi_type() for setinpusizes

Adjusted the "setinputsizes" logic relied upon by the cx_Oracle, asyncpg
and pg8000 dialects to support a :class:`.TypeDecorator` that includes
an override the :meth:`.TypeDecorator.get_dbapi_type()` method.

Change-Id: I5aa70abf0d9a9e2ca43309f2dd80b3fcd83881b9

2 files changed, 48 insertions, 3 deletions

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index a734bb582..8ee575cca 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1047,14 +1047,19 @@ class SQLCompiler(Compiled):
         if include_types is None and exclude_types is None:
 
             def _lookup_type(typ):
-                dialect_impl = typ._unwrapped_dialect_impl(dialect)
-                return dialect_impl.get_dbapi_type(dbapi)
+                dbtype = typ.dialect_impl(dialect).get_dbapi_type(dbapi)
+                return dbtype
 
         else:
 
             def _lookup_type(typ):
+                # note we get dbtype from the possibly TypeDecorator-wrapped
+                # dialect_impl, but the dialect_impl itself that we use for
+                # include/exclude is the unwrapped version.
+
                 dialect_impl = typ._unwrapped_dialect_impl(dialect)
-                dbtype = dialect_impl.get_dbapi_type(dbapi)
+
+                dbtype = typ.dialect_impl(dialect).get_dbapi_type(dbapi)
 
                 if (
                     dbtype is not None
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 749e83de4..43777239c 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -462,6 +462,45 @@ class IntegerTest(_LiteralRoundTripFixture, fixtures.TestBase):
             assert isinstance(row[0], (long, int))  # noqa
 
 
+class CastTypeDecoratorTest(_LiteralRoundTripFixture, fixtures.TestBase):
+    __backend__ = True
+
+    @testing.fixture
+    def string_as_int(self):
+        class StringAsInt(TypeDecorator):
+            impl = String(50)
+
+            def get_dbapi_type(self, dbapi):
+                return dbapi.NUMBER
+
+            def column_expression(self, col):
+                return cast(col, Integer)
+
+            def bind_expression(self, col):
+                return cast(col, String(50))
+
+        return StringAsInt()
+
+    @testing.provide_metadata
+    def test_special_type(self, connection, string_as_int):
+
+        type_ = string_as_int
+
+        metadata = self.metadata
+        t = Table("t", metadata, Column("x", type_))
+        t.create(connection)
+
+        connection.execute(t.insert(), [{"x": x} for x in [1, 2, 3]])
+
+        result = {row[0] for row in connection.execute(t.select())}
+        eq_(result, {1, 2, 3})
+
+        result = {
+            row[0] for row in connection.execute(t.select().where(t.c.x == 2))
+        }
+        eq_(result, {2})
+
+
 class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase):
     __backend__ = True
 
@@ -1302,6 +1341,7 @@ __all__ = (
     "TextTest",
     "NumericTest",
     "IntegerTest",
+    "CastTypeDecoratorTest",
     "DateTimeHistoricTest",
     "DateTimeCoercedToDateTimeTest",
     "TimeMicrosecondsTest",