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

3 files changed, 53 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_13/5170.rst b/doc/build/changelog/unreleased_13/5170.rst
new file mode 100644
index 000000000..2cebdd655
--- /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 0a442f256..bfe3812be 100644
--- 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 830a54eef..548338e9e 100644
--- 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(