Use pg_get_constraintdef for CHECK constraint reflection

Revised the query used when reflecting CHECK constraints to make use of the
``pg_get_constraintdef`` function, as the ``consrc`` column is being
deprecated in PG 12.  Thanks to John A Stevenson for the tip.

Fixes: #4463
Change-Id: Ie0ee9bdfddb0635db72b35c2e2e4b27f154162fd

diff --git a/doc/build/changelog/unreleased_12/4463.rst b/doc/build/changelog/unreleased_12/4463.rst
new file mode 100644 (file)
index 0000000..a8baa82
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4463.rst
@@ -0,0 +1,8 @@
+.. change::
+   :tags: bug, postgresql
+   :tickets: 4463
+
+   Revised the query used when reflecting CHECK constraints to make use of the
+   ``pg_get_constraintdef`` function, as the ``consrc`` column is being
+   deprecated in PG 12.  Thanks to John A Stevenson for the tip.
+

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index ecdbe001814211c9603a8125168e455b49c39578..e9040fb4349975acb00b511bf80826e99c4e90e7 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3381,7 +3381,7 @@ class PGDialect(default.DefaultDialect):
         CHECK_SQL = """
             SELECT
                 cons.conname as name,
-                cons.consrc as src
+                pg_get_constraintdef(cons.oid) as src
             FROM
                 pg_catalog.pg_constraint cons
             WHERE
@@ -3391,8 +3391,19 @@ class PGDialect(default.DefaultDialect):
 
         c = connection.execute(sql.text(CHECK_SQL), table_oid=table_oid)
 
+        # samples:
+        # "CHECK (((a > 1) AND (a < 5)))"
+        # "CHECK (((a = 1) OR ((a > 2) AND (a < 5))))"
+        def match_cons(src):
+            m = re.match(r"^CHECK *\(\((.+)\)\)$", src)
+            if not m:
+                util.warn("Could not parse CHECK constraint text: %r" % src)
+                return ""
+            return m.group(1)
+
         return [
-            {"name": name, "sqltext": src[1:-1]} for name, src in c.fetchall()
+            {"name": name, "sqltext": match_cons(src)}
+            for name, src in c.fetchall()
         ]
 
     def _load_enums(self, connection, schema=None):

diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 311835daee045ab380a607e0b597c317a489d93e..ae1dff6d058443522dee8b684d36a1b0cadf4fd8 100644 (file)
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -28,6 +28,7 @@ from sqlalchemy.dialects.postgresql import TSRANGE
 from sqlalchemy.engine import reflection
 from sqlalchemy.sql.schema import CheckConstraint
 from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import mock
 from sqlalchemy.testing.assertions import assert_raises
 from sqlalchemy.testing.assertions import AssertsExecutionResults
 from sqlalchemy.testing.assertions import eq_
@@ -1424,6 +1425,20 @@ class ReflectionTest(fixtures.TestBase):
             },
         )
 
+    def test_reflect_check_warning(self):
+        conn = mock.Mock(
+            execute=lambda *arg, **kw: mock.Mock(
+                fetchall=lambda: [("some name", "NOTCHECK foobar")]
+            )
+        )
+        with mock.patch.object(
+            testing.db.dialect, "get_table_oid", lambda *arg, **kw: 1
+        ):
+            with testing.expect_warnings(
+                "Could not parse CHECK constraint text: 'NOTCHECK foobar'"
+            ):
+                testing.db.dialect.get_check_constraints(conn, "foo")
+
 
 class CustomTypeReflectionTest(fixtures.TestBase):
     class CustomType(object):