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

1 files changed, 31 insertions, 0 deletions

diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index f0893d822..481924c38 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -43,9 +43,11 @@ from sqlalchemy.testing.assertions import AssertsExecutionResults
 from sqlalchemy.testing.assertions import ComparesIndexes
 from sqlalchemy.testing.assertions import eq_
 from sqlalchemy.testing.assertions import expect_raises
+from sqlalchemy.testing.assertions import expect_warnings
 from sqlalchemy.testing.assertions import is_
 from sqlalchemy.testing.assertions import is_false
 from sqlalchemy.testing.assertions import is_true
+from sqlalchemy.types import NullType
 
 
 class ReflectionFixtures:
@@ -2305,6 +2307,35 @@ 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'"
+        ):
+            row_dict = {
+                "name": "colname",
+                "table_name": "tblname",
+                "format_type": None,
+                "default": None,
+                "not_null": False,
+                "comment": None,
+                "generated": "",
+                "identity_options": None,
+            }
+            column_info = dialect._get_columns_info(
+                [row_dict], {}, {}, "public"
+            )
+            assert ("public", "tblname") in column_info
+            column_info = column_info[("public", "tblname")]
+            assert len(column_info) == 1
+            column_info = column_info[0]
+            assert isinstance(column_info["type"], NullType)
+
 
 class IntervalReflectionTest(fixtures.TestBase):
     __only_on__ = "postgresql"