--- /dev/null
+.. change::
+ :tags: bug, postgresql, reflection
+ :tickets: 13157
+
+ Fixed reflection of PostgreSQL CHECK constraints where an expression made
+ up of multiple parenthesized sub-expressions, such as ``(x IS NULL OR y IS
+ NULL) AND (x IS NULL OR y IS NULL)``, would have its leading and trailing
+ parentheses incorrectly stripped, producing an unbalanced and
+ syntactically invalid reflected expression. Pull request courtesy
+ Shaurya Singh.
util.warn("Could not parse CHECK constraint text: %r" % src)
sqltext = ""
else:
- sqltext = re.compile(
- r"^[\s\n]*\((.+)\)[\s\n]*$", flags=re.DOTALL
- ).sub(r"\1", m.group(1))
+ sqltext = util.strip_outer_parens(m.group(1))
entry = {
"name": check_name,
"sqltext": sqltext,
flags=re.DOTALL,
)
- # Find the matching closing parenthesis by counting balanced parens
- # Must track string context to ignore parens inside string literals
- start = match.end() # Position after 'CHECK ('
- paren_count = 1
- in_single_quote = False
- in_double_quote = False
-
- for pos, char in enumerate(table_data[start:], start):
- # Track string literal context
- if char == "'" and not in_double_quote:
- in_single_quote = not in_single_quote
- elif char == '"' and not in_single_quote:
- in_double_quote = not in_double_quote
- # Only count parens when not inside a string literal
- elif not in_single_quote and not in_double_quote:
- if char == "(":
- paren_count += 1
- elif char == ")":
- paren_count -= 1
- if paren_count == 0:
- # Successfully found matching closing parenthesis
- sqltext = table_data[start:pos].strip()
- cks.append(
- {"sqltext": sqltext, "name": constraint_name}
- )
- break
+ # Find the matching closing parenthesis with quote-aware paren
+ # counting. ``match.end() - 1`` is the position of the ``(``
+ # that opened the CHECK clause; ``match.end()`` is the first
+ # character of the constraint body.
+ close = util.find_matching_paren(table_data, match.end() - 1)
+ if close is not None:
+ sqltext = table_data[match.end() : close].strip()
+ cks.append({"sqltext": sqltext, "name": constraint_name})
cks.sort(key=lambda d: d["name"] or "~") # sort None as last
if cks:
],
)
+ def _cc_by_name(self, reflected, name):
+ """return the lower cased sqltext for the named CHECK constraint."""
+
+ for rec in reflected:
+ if rec["name"] == name:
+ return rec["sqltext"].lower()
+
+ assert False, (
+ f"No CHECK constraint named {name!r} in "
+ f"{[rec['name'] for rec in reflected]}"
+ )
+
+ @testing.requires.check_constraint_reflection
+ @testing.combinations(
+ # Regression for #13157: independent sibling parens must not be
+ # treated as if they wrap the whole expression.
+ "(x IS NULL OR y IS NULL) AND (x IS NULL OR y IS NULL)",
+ # Parentheses inside string literals must not throw off the
+ # paren counter.
+ "a = '(' AND b = ')'",
+ # Doubled '' inside a quoted literal is the SQL single-quote
+ # escape.
+ "a = 'it''s'",
+ # Redundant nested parens around a boolean expression.
+ "((((x > 0))))",
+ # Parenthesized sub-expressions.
+ "((x > 1) AND (x < 5))",
+ # Plain expression, no outer parens.
+ "x > 0",
+ argnames="expression",
+ )
+ def test_check_constraint_parenthesized_expressions(
+ self, metadata, inspect_for_table, expression
+ ):
+ """Regression test for #13157.
+
+ A CHECK constraint expression must round-trip through reflection
+ without its parentheses being incorrectly stripped. The bug
+ greedily paired the leading ``(`` with the trailing ``)`` and
+ dropped both, producing an unbalanced, syntactically invalid
+ expression such as ``x IS NULL OR y IS NULL) AND (x IS NULL OR
+ y IS NULL``.
+ """
+ with inspect_for_table("sa_cc") as (schema, inspector):
+ Table(
+ "sa_cc",
+ metadata,
+ Column("id", Integer(), primary_key=True),
+ Column("x", Integer()),
+ Column("y", Integer()),
+ Column("a", String(50)),
+ Column("b", String(50)),
+ sa.CheckConstraint(expression, name="cc_expr"),
+ schema=schema,
+ )
+
+ reflected = inspector.get_check_constraints("sa_cc", schema=schema)
+
+ # some DBs like Oracle may create additional CHECK constraints
+ # implicitly, so locate ours by name
+
+ reflected_text = self._cc_by_name(reflected, "cc_expr")
+
+ # since different DBs normalize differently, e.g. postgresql
+ # collapses redundant parens, Oracle returns the whole expression
+ # inside of additional parens, MySQL has different quotes, etc.
+ # create a new table + CHECK constraint with our reflected text,
+ # then assert that this new constraint reflects identically to the
+ # original, proving that the database represents both the original
+ # constraint and the reflected text identically.
+ with inspect_for_table("sa_cc_2") as (schema, inspector):
+ Table(
+ "sa_cc_2",
+ metadata,
+ Column("id", Integer(), primary_key=True),
+ Column("x", Integer()),
+ Column("y", Integer()),
+ Column("a", String(50)),
+ Column("b", String(50)),
+ sa.CheckConstraint(reflected_text, name="cc_expr_2"),
+ schema=schema,
+ )
+
+ reflected2 = inspector.get_check_constraints("sa_cc_2", schema=schema)
+
+ reflected_text_2 = self._cc_by_name(reflected2, "cc_expr_2")
+ eq_(reflected_text, reflected_text_2)
+
@testing.requires.indexes_check_column_order
def test_index_column_order(self, metadata, inspect_for_table):
"""test for #12894"""
from .langhelpers import ellipses_string as ellipses_string
from .langhelpers import EnsureKWArg as EnsureKWArg
from .langhelpers import FastIntFlag as FastIntFlag
+from .langhelpers import find_matching_paren as find_matching_paren
from .langhelpers import format_argspec_init as format_argspec_init
from .langhelpers import format_argspec_plus as format_argspec_plus
from .langhelpers import generic_fn_descriptor as generic_fn_descriptor
from .langhelpers import safe_reraise as safe_reraise
from .langhelpers import set_creation_order as set_creation_order
from .langhelpers import string_or_unprintable as string_or_unprintable
+from .langhelpers import strip_outer_parens as strip_outer_parens
from .langhelpers import symbol as symbol
from .langhelpers import TypingOnly as TypingOnly
from .langhelpers import (
return _f
+def find_matching_paren(text: str, start: int = 0) -> Optional[int]:
+ """Return the index of the ``)`` that matches the ``(`` at ``start``.
+
+ The walk skips single-quoted (``'...'``) and double-quoted (``"..."``)
+ string literals, so parentheses inside string literals do not affect
+ the depth counter. ``''`` and ``""`` are treated as escaped quotes
+ inside their respective contexts, matching PostgreSQL/SQLite literal
+ conventions.
+
+ Returns ``None`` if the opening parenthesis is never closed
+ (unbalanced). The character at ``text[start]`` must be ``(``.
+
+ Note for SQLite use, SQLite also supports MySQL backtick-style quotes as
+ well as SQL Server bracket style quotes; the latter has different escaping
+ behaviors. A follow-up patch could add support for these two additional
+ styles (consider using an enum like QuotingStyle.DOUBLE |
+ QuotingStyle.BRACKET, etc.)
+
+ E.g.::
+
+ >>> find_matching_paren("(a + b)")
+ 6
+ >>> find_matching_paren("((a)(b))")
+ 7
+ >>> find_matching_paren("(a = '(' AND b = ')')")
+ 20
+
+ """
+ assert text[start] == "(", "start index must point at an open paren"
+
+ depth = 0
+ in_single = False
+ in_double = False
+ n = len(text)
+ i = start
+ while i < n:
+ ch = text[i]
+ if in_single:
+ if ch == "'":
+ if i + 1 < n and text[i + 1] == "'":
+ i += 2
+ continue
+ in_single = False
+ elif in_double:
+ if ch == '"':
+ if i + 1 < n and text[i + 1] == '"':
+ i += 2
+ continue
+ in_double = False
+ elif ch == "'":
+ in_single = True
+ elif ch == '"':
+ in_double = True
+ elif ch == "(":
+ depth += 1
+ elif ch == ")":
+ depth -= 1
+ if depth == 0:
+ return i
+ i += 1
+ return None
+
+
+def strip_outer_parens(text: str) -> str:
+ """Remove one layer of outer parentheses from ``text`` if they wrap the
+ entire (stripped) string.
+
+ Whitespace is preserved if the parentheses do not wrap the whole
+ expression. String literals are honored via :func:`find_matching_paren`,
+ so ``"(a = '(' AND b = ')')"`` correctly strips to
+ ``"a = '(' AND b = ')'"`` rather than being interpreted as two separate
+ paren groups.
+
+ E.g.::
+
+ >>> strip_outer_parens("(a IS NOT NULL)")
+ 'a IS NOT NULL'
+ >>> strip_outer_parens("(a) AND (b)")
+ '(a) AND (b)'
+ >>> strip_outer_parens("a NOT NULL")
+ 'a NOT NULL'
+
+ """
+ stripped = text.strip()
+ lstripped = len(stripped)
+ if lstripped < 2 or stripped[0] != "(" or stripped[-1] != ")":
+ return text
+ close = find_matching_paren(stripped, 0)
+ if close is not None and close == lstripped - 1:
+ return stripped[1:-1]
+ return text
+
+
def quoted_token_parser(value):
"""Parse a dotted identifier with accommodation for quoted names.
eq_(util.parse_version_from_metadata("no_such_distribution"), ())
+class ParenthesisBalancingTest(fixtures.TestBase):
+ """test the parenthesis functions added as part of #13157"""
+
+ @testing.combinations(
+ ("(a + b)", 0, 6),
+ ("((a)(b))", 0, 7),
+ ("(a IS NULL OR b IS NULL)", 0, 23),
+ # Sibling parens, not nested -- depth returns to 0 before the end.
+ ("(a) AND (b)", 0, 2),
+ # Parens inside single-quoted literal must not affect the count.
+ ("(a = '(' AND b = ')')", 0, 20),
+ # Doubled single-quote inside a quoted literal is a single-quote
+ # escape; the literal continues until the next unescaped ``'``.
+ ("(a = 'it''s')", 0, 12),
+ # Parens inside double-quoted identifier must not affect the count.
+ ('("col(1)" IS NOT NULL)', 0, 21),
+ # ``start`` other than 0.
+ ("xx(a)yy", 2, 4),
+ argnames="text, start, expected",
+ )
+ def test_balanced(self, text, start, expected):
+ eq_(langhelpers.find_matching_paren(text, start), expected)
+
+ def test_unbalanced(self):
+ eq_(langhelpers.find_matching_paren("(a + b", 0), None)
+ eq_(langhelpers.find_matching_paren("(a + b'", 0), None)
+
+ def test_start_not_open_paren(self):
+ with expect_raises(AssertionError):
+ langhelpers.find_matching_paren("a + b", 0)
+
+ @testing.combinations(
+ ("(a IS NOT NULL)", "a IS NOT NULL"),
+ ("((a > 1) AND (a < 5))", "(a > 1) AND (a < 5)"),
+ # Multiply-wrapped: strip only the outer layer.
+ ("(((a > 1) AND (a < 5)))", "((a > 1) AND (a < 5))"),
+ # Already unwrapped -- no change.
+ ("a NOT NULL", "a NOT NULL"),
+ # Sibling parens, NOT a single wrap -- must not strip.
+ (
+ "(x IS NULL OR y IS NULL) AND (x IS NULL OR y IS NULL)",
+ "(x IS NULL OR y IS NULL) AND (x IS NULL OR y IS NULL)",
+ ),
+ # Parens inside literals must not throw off the depth counter.
+ ("(a = '(' AND b = ')')", "a = '(' AND b = ')'"),
+ ("a = '(' AND b = ')'", "a = '(' AND b = ')'"),
+ # Whitespace is preserved on the unstripped portion.
+ (
+ "(\n(a < 1)\n OR\n (a >= 5)\n)",
+ "\n(a < 1)\n OR\n (a >= 5)\n",
+ ),
+ # Empty / pathological inputs.
+ ("", ""),
+ ("()", ""),
+ ("(", "("),
+ (")", ")"),
+ argnames="src, expected",
+ )
+ def test_strip(self, src, expected):
+ eq_(langhelpers.strip_outer_parens(src), expected)
+
+
class BackslashReplaceTest(fixtures.TestBase):
def test_ascii_to_utf8(self):
eq_(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
