accommodate NULL format_type()

Made an adjustment to how the PostgreSQL dialect considers column types
when it reflects columns from a table, to accommodate for alternative
backends which may return NULL from the PG ``format_type()`` function.

Fixes: #8748
Change-Id: I6178287aac567210a76afaa5805b825daa7fa4db
(cherry picked from commit 200e70b9745f1f344be4a35bb8f2b5f01b40d467)

diff --git a/doc/build/changelog/unreleased_14/8748.rst b/doc/build/changelog/unreleased_14/8748.rst
new file mode 100644 (file)
index 0000000..27e0679
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8748.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug postgresql
+    :tickets: 8748
+
+    Made an adjustment to how the PostgreSQL dialect considers column types
+    when it reflects columns from a table, to accommodate for alternative
+    backends which may return NULL from the PG ``format_type()`` function.

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index b980183d007bd2f764888423bfc800759cdcd108..c390553353a68ecb5bffa7c8b7aa17b1f81e56ea 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3975,12 +3975,19 @@ class PGDialect(default.DefaultDialect):
                 attype.endswith("[]"),
             )
 
-        # strip (*) from character varying(5), timestamp(5)
-        # with time zone, geometry(POLYGON), etc.
-        attype = re.sub(r"\(.*\)", "", format_type)
+        if format_type is None:
+            no_format_type = True
+            attype = format_type = "no format_type()"
+            is_array = False
+        else:
+            no_format_type = False
+
+            # strip (*) from character varying(5), timestamp(5)
+            # with time zone, geometry(POLYGON), etc.
+            attype = re.sub(r"\(.*\)", "", format_type)
 
-        # strip '[]' from integer[], etc. and check if an array
-        attype, is_array = _handle_array_type(attype)
+            # strip '[]' from integer[], etc. and check if an array
+            attype, is_array = _handle_array_type(attype)
 
         # strip quotes from case sensitive enum or domain names
         enum_or_domain_key = tuple(util.quoted_token_parser(attype))
@@ -4073,6 +4080,12 @@ class PGDialect(default.DefaultDialect):
             coltype = coltype(*args, **kwargs)
             if is_array:
                 coltype = self.ischema_names["_array"](coltype)
+        elif no_format_type:
+            util.warn(
+                "PostgreSQL format_type() returned NULL for column '%s'"
+                % (name,)
+            )
+            coltype = sqltypes.NULLTYPE
         else:
             util.warn(
                 "Did not recognize type '%s' of column '%s'" % (attype, name)

diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index bf8cd511116f2fa7705f413b36ee740276f9280a..f33b3bde45417af9c36e18e51d50d38a36a59fd4 100644 (file)
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -37,8 +37,10 @@ from sqlalchemy.testing.assertions import assert_raises
 from sqlalchemy.testing.assertions import assert_warns
 from sqlalchemy.testing.assertions import AssertsExecutionResults
 from sqlalchemy.testing.assertions import eq_
+from sqlalchemy.testing.assertions import expect_warnings
 from sqlalchemy.testing.assertions import is_
 from sqlalchemy.testing.assertions import is_true
+from sqlalchemy.types import NullType
 
 
 class ReflectionFixtures(object):
@@ -1822,6 +1824,21 @@ class CustomTypeReflectionTest(fixtures.TestBase):
         dialect.ischema_names["my_custom_type"] = self.CustomType
         self._assert_reflected(dialect)
 
+    def test_no_format_type(self):
+        """test #8748"""
+
+        dialect = postgresql.PGDialect()
+        dialect.ischema_names = dialect.ischema_names.copy()
+        dialect.ischema_names["my_custom_type"] = self.CustomType
+
+        with expect_warnings(
+            r"PostgreSQL format_type\(\) returned NULL for column 'colname'"
+        ):
+            column_info = dialect._get_column_info(
+                "colname", None, None, False, {}, {}, "public", None, "", None
+            )
+            assert isinstance(column_info["type"], NullType)
+
 
 class IntervalReflectionTest(fixtures.TestBase):
     __only_on__ = "postgresql"