--- /dev/null
+.. change::
+ :tags: bug, engine
+ :tickets: 10275
+
+ Fixed a series of reflection issues affecting the PostgreSQL,
+ MySQL/MariaDB, and SQLite dialects when reflecting foreign key constraints
+ where the target column contained parenthesis in one or both of the table
+ name or column name.
+
r"\((?P<local>[^\)]+?)\) REFERENCES +"
r"(?P<table>%(iq)s[^%(fq)s]+%(fq)s"
r"(?:\.%(iq)s[^%(fq)s]+%(fq)s)?) +"
- r"\((?P<foreign>[^\)]+?)\)"
+ r"\((?P<foreign>(?:%(iq)s[^%(fq)s]+%(fq)s(?: *, *)?)+)\)"
r"(?: +(?P<match>MATCH \w+))?"
r"(?: +ON DELETE (?P<ondelete>%(on)s))?"
r"(?: +ON UPDATE (?P<onupdate>%(on)s))?" % kw
@util.memoized_property
def _fk_regex_pattern(self):
+ # optionally quoted token
+ qtoken = '(?:"[^"]+"|[A-Za-z0-9_]+?)'
+
# https://www.postgresql.org/docs/current/static/sql-createtable.html
return re.compile(
- r"FOREIGN KEY \((.*?)\) REFERENCES (?:(.*?)\.)?(.*?)\((.*?)\)"
+ r"FOREIGN KEY \((.*?)\) "
+ rf"REFERENCES (?:({qtoken})\.)?({qtoken})\(((?:{qtoken}(?: *, *)?)+)\)" # noqa: E501
r"[\s]?(MATCH (FULL|PARTIAL|SIMPLE)+)?"
r"[\s]?(ON UPDATE "
r"(CASCADE|RESTRICT|NO ACTION|SET NULL|SET DEFAULT)+)?"
if table_data is None:
# system tables, etc.
return
+
+ # note that we already have the FKs from PRAGMA above. This whole
+ # regexp thing is trying to locate additional detail about the
+ # FKs, namely the name of the constraint and other options.
+ # so parsing the columns is really about matching it up to what
+ # we already have.
FK_PATTERN = (
r"(?:CONSTRAINT (\w+) +)?"
r"FOREIGN KEY *\( *(.+?) *\) +"
- r'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\((.+?)\) *'
+ r'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\( *((?:(?:"[^"]+"|[a-z0-9_]+) *(?:, *)?)+)\) *' # noqa: E501
r"((?:ON (?:DELETE|UPDATE) "
r"(?:SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION) *)*)"
r"((?:NOT +)?DEFERRABLE)?"
)
+class BizarroCharacterFKResolutionTest(fixtures.TestBase):
+ """tests for #10275"""
+
+ __backend__ = True
+
+ @testing.combinations(
+ ("id",), ("(3)",), ("col%p",), ("[brack]",), argnames="columnname"
+ )
+ @testing.variation("use_composite", [True, False])
+ @testing.combinations(
+ ("plain",),
+ ("(2)",),
+ ("per % cent",),
+ ("[brackets]",),
+ argnames="tablename",
+ )
+ def test_fk_ref(
+ self, connection, metadata, use_composite, tablename, columnname
+ ):
+ tt = Table(
+ tablename,
+ metadata,
+ Column(columnname, Integer, key="id", primary_key=True),
+ test_needs_fk=True,
+ )
+ if use_composite:
+ tt.append_column(Column("id2", Integer, primary_key=True))
+
+ if use_composite:
+ Table(
+ "other",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("ref", Integer),
+ Column("ref2", Integer),
+ sa.ForeignKeyConstraint(["ref", "ref2"], [tt.c.id, tt.c.id2]),
+ test_needs_fk=True,
+ )
+ else:
+ Table(
+ "other",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("ref", ForeignKey(tt.c.id)),
+ test_needs_fk=True,
+ )
+
+ metadata.create_all(connection)
+
+ m2 = MetaData()
+
+ o2 = Table("other", m2, autoload_with=connection)
+ t1 = m2.tables[tablename]
+
+ assert o2.c.ref.references(t1.c[0])
+ if use_composite:
+ assert o2.c.ref2.references(t1.c[1])
+
+
class QuotedNameArgumentTest(fixtures.TablesTest):
run_create_tables = "once"
__backend__ = True
"ComponentReflectionTestExtra",
"TableNoColumnsTest",
"QuotedNameArgumentTest",
+ "BizarroCharacterFKResolutionTest",
"HasTableTest",
"HasIndexTest",
"NormalizedNameTest",
import datetime
import logging
import logging.handlers
+import re
from sqlalchemy import BigInteger
from sqlalchemy import bindparam
from sqlalchemy.testing.assertions import ne_
+def _fk_expected(
+ constrained_columns,
+ referred_table,
+ referred_columns,
+ referred_schema=None,
+ match=None,
+ onupdate=None,
+ ondelete=None,
+ deferrable=None,
+ initially=None,
+):
+ return (
+ constrained_columns,
+ referred_schema,
+ referred_table,
+ referred_columns,
+ f"MATCH {match}" if match else None,
+ match,
+ f"ON UPDATE {onupdate}" if onupdate else None,
+ onupdate,
+ f"ON DELETE {ondelete}" if ondelete else None,
+ ondelete,
+ deferrable,
+ f"INITIALLY {initially}" if initially else None,
+ initially,
+ )
+
+
class DialectTest(fixtures.TestBase):
"""python-side dialect tests."""
+ @testing.combinations(
+ (
+ "FOREIGN KEY (tid) REFERENCES some_table(id)",
+ _fk_expected("tid", "some_table", "id"),
+ ),
+ (
+ 'FOREIGN KEY (tid) REFERENCES "(2)"(id)',
+ _fk_expected("tid", '"(2)"', "id"),
+ ),
+ (
+ 'FOREIGN KEY (tid) REFERENCES some_table("(2)")',
+ _fk_expected("tid", "some_table", '"(2)"'),
+ ),
+ (
+ 'FOREIGN KEY (tid1, tid2) REFERENCES some_table("(2)", "(3)")',
+ _fk_expected("tid1, tid2", "some_table", '"(2)", "(3)"'),
+ ),
+ (
+ "FOREIGN KEY (tid) REFERENCES some_table(id) "
+ "DEFERRABLE INITIALLY DEFERRED",
+ _fk_expected(
+ "tid",
+ "some_table",
+ "id",
+ deferrable="DEFERRABLE",
+ initially="DEFERRED",
+ ),
+ ),
+ (
+ "FOREIGN KEY (tid1, tid2) "
+ "REFERENCES some_schema.some_table(id1, id2)",
+ _fk_expected(
+ "tid1, tid2",
+ "some_table",
+ "id1, id2",
+ referred_schema="some_schema",
+ ),
+ ),
+ (
+ "FOREIGN KEY (tid1, tid2) "
+ "REFERENCES some_schema.some_table(id1, id2) "
+ "MATCH FULL "
+ "ON UPDATE CASCADE "
+ "ON DELETE CASCADE "
+ "DEFERRABLE INITIALLY DEFERRED",
+ _fk_expected(
+ "tid1, tid2",
+ "some_table",
+ "id1, id2",
+ referred_schema="some_schema",
+ onupdate="CASCADE",
+ ondelete="CASCADE",
+ match="FULL",
+ deferrable="DEFERRABLE",
+ initially="DEFERRED",
+ ),
+ ),
+ )
+ def test_fk_parsing(self, condef, expected):
+ FK_REGEX = postgresql.dialect()._fk_regex_pattern
+ groups = re.search(FK_REGEX, condef).groups()
+
+ eq_(groups, expected)
+
def test_range_constructor(self):
"""test kwonly argments in the range constructor, as we had
to do dataclasses backwards compat operations"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
