While parsing for check constraints, ignore newline characters in the check condition

Fixed bug where PostgreSQL reflection of CHECK constraints would fail to
parse the constraint if the SQL text contained newline characters. The
regular expression has been adjusted to accommodate for this case. Pull
request courtesy Eric Borczuk.

Fixes: #5170
Closes: #5172
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/5172
Pull-request-sha: 5701b7f09f723b727bbee95d19d107d6cc1d7717

Change-Id: If727e9140b645e8b685c3476fb0fa4417c1e6526
(cherry picked from commit 0fe528483afeb53300ff7a9770f7fb9c81a3a874)

diff --git a/doc/build/changelog/unreleased_13/5170.rst b/doc/build/changelog/unreleased_13/5170.rst
new file mode 100644 (file)
index 0000000..2cebdd6
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5170.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, postgresql, reflection
+    :tickets: 5170
+
+    Fixed bug where PostgreSQL reflection of CHECK constraints would fail to
+    parse the constraint if the SQL text contained newline characters. The
+    regular expression has been adjusted to accommodate for this case. Pull
+    request courtesy Eric Borczuk.

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 5f5ea8826e84b46f59c3fe9b449574190b227f01..ff0e0535e4ff1f9afe3cdfb551ad473d198fed4e 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3489,12 +3489,18 @@ class PGDialect(default.DefaultDialect):
             # "CHECK (((a = 1) OR ((a > 2) AND (a < 5))))"
             # "CHECK (((a > 1) AND (a < 5))) NOT VALID"
             # "CHECK (some_boolean_function(a))"
-            m = re.match(r"^CHECK *\((.+)\)( NOT VALID)?$", src)
+            # "CHECK (((a\n < 1)\n OR\n (a\n >= 5))\n)"
+
+            m = re.match(
+                r"^CHECK *\((.+)\)( NOT VALID)?$", src, flags=re.DOTALL
+            )
             if not m:
                 util.warn("Could not parse CHECK constraint text: %r" % src)
                 sqltext = ""
             else:
-                sqltext = re.sub(r"^\((.+)\)$", r"\1", m.group(1))
+                sqltext = re.compile(
+                    r"^[\s\n]*\((.+)\)[\s\n]*$", flags=re.DOTALL
+                ).sub(r"\1", m.group(1))
             entry = {"name": name, "sqltext": sqltext}
             if m and m.group(2):
                 entry["dialect_options"] = {"not_valid": True}

diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 86d0f3e38d29b9344daeda34d11dbeffa73f3ef0..e7297e236433efe3cfd5c11a411f0b6fda325a09 100644 (file)
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -1550,9 +1550,11 @@ class ReflectionTest(fixtures.TestBase):
             "pgsql_cc",
             meta,
             Column("a", Integer()),
+            Column("b", String),
             CheckConstraint("a > 1 AND a < 5", name="cc1"),
             CheckConstraint("a = 1 OR (a > 2 AND a < 5)", name="cc2"),
             CheckConstraint("is_positive(a)", name="cc3"),
+            CheckConstraint("b != 'hi\nim a name   \nyup\n'", name="cc4"),
         )
 
         meta.create_all()
@@ -1571,6 +1573,7 @@ class ReflectionTest(fixtures.TestBase):
                 u"cc1": u"(a > 1) AND (a < 5)",
                 u"cc2": u"(a = 1) OR ((a > 2) AND (a < 5))",
                 u"cc3": u"is_positive(a)",
+                u"cc4": u"(b)::text <> 'hi\nim a name   \nyup\n'::text",
             },
         )
 
@@ -1589,6 +1592,40 @@ class ReflectionTest(fixtures.TestBase):
             ):
                 testing.db.dialect.get_check_constraints(conn, "foo")
 
+    def test_reflect_extra_newlines(self):
+        rows = [
+            ("some name", "CHECK (\n(a \nIS\n NOT\n\n NULL\n)\n)"),
+            ("some other name", "CHECK ((b\nIS\nNOT\nNULL))"),
+            ("some CRLF name", "CHECK ((c\r\n\r\nIS\r\nNOT\r\nNULL))"),
+            ("some name", "CHECK (c != 'hi\nim a name\n')"),
+        ]
+        conn = mock.Mock(
+            execute=lambda *arg, **kw: mock.MagicMock(
+                fetchall=lambda: rows, __iter__=lambda self: iter(rows)
+            )
+        )
+        with mock.patch.object(
+            testing.db.dialect, "get_table_oid", lambda *arg, **kw: 1
+        ):
+            check_constraints = testing.db.dialect.get_check_constraints(
+                conn, "foo"
+            )
+            eq_(
+                check_constraints,
+                [
+                    {
+                        "name": "some name",
+                        "sqltext": "a \nIS\n NOT\n\n NULL\n",
+                    },
+                    {"name": "some other name", "sqltext": "b\nIS\nNOT\nNULL"},
+                    {
+                        "name": "some CRLF name",
+                        "sqltext": "c\r\n\r\nIS\r\nNOT\r\nNULL",
+                    },
+                    {"name": "some name", "sqltext": "c != 'hi\nim a name\n'"},
+                ],
+            )
+
     def test_reflect_with_not_valid_check_constraint(self):
         rows = [("some name", "CHECK ((a IS NOT NULL)) NOT VALID")]
         conn = mock.Mock(