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

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 0a442f2562224ccd8cad22a7fd84fe75fdf312c9..bfe3812bed0403ffa9825ec25c95cb82e9e66649 100644 (file)
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3488,12 +3488,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 830a54eef0dad1759a972b3e9fd6b553b07f3a44..548338e9e4fb77e26f24096cccae8d3b9bd3cb4b 100644 (file)
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -1543,9 +1543,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()
@@ -1564,6 +1566,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",
             },
         )
 
@@ -1582,6 +1585,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(