From: Mike Bayer Date: Sun, 20 Feb 2022 14:51:22 +0000 (-0500) Subject: improve reflection of inline UNIQUE constraints X-Git-Tag: rel_1_4_32~18^2 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=9d0a17ad27e9eab14bd52ec54fbfb4803c74a535;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git improve reflection of inline UNIQUE constraints Fixed issue where SQLite unique constraint reflection would not work for an inline UNIQUE constraint where the column name had an underscore in its name. Added support for reflecting SQLite inline unique constraints where the column names are formatted with SQLite "escape quotes" ``[]`` or `` ` ``, which are discarded by the database when producing the column name. Fixes: #7736 Change-Id: I635003478dc27193995f7d7a6448f9333a498706 (cherry picked from commit 834af17a469fd1893acf20225e8400c0c908053f) --- diff --git a/doc/build/changelog/unreleased_14/7736.rst b/doc/build/changelog/unreleased_14/7736.rst new file mode 100644 index 0000000000..828dd540a4 --- /dev/null +++ b/doc/build/changelog/unreleased_14/7736.rst @@ -0,0 +1,16 @@ +.. change:: + :tags: bug, sqlite + :tickets: 7736 + + Fixed issue where SQLite unique constraint reflection would not work + for an inline UNIQUE constraint where the column name had an underscore + in its name. + +.. change:: + :tags: usecase, sqlite + :tickets: 7736 + + Added support for reflecting SQLite inline unique constraints where + the column names are formatted with SQLite "escape quotes" ``[]`` + or `` ` ``, which are discarded by the database when producing the + column name. diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py index bcea17620f..7ba9700d70 100644 --- a/lib/sqlalchemy/dialects/sqlite/base.py +++ b/lib/sqlalchemy/dialects/sqlite/base.py @@ -2414,7 +2414,8 @@ class SQLiteDialect(default.DefaultDialect): def parse_uqs(): UNIQUE_PATTERN = r'(?:CONSTRAINT "?(.+?)"? +)?UNIQUE *\((.+?)\)' INLINE_UNIQUE_PATTERN = ( - r'(?:(".+?")|([a-z0-9]+)) ' r"+[a-z0-9_ ]+? +UNIQUE" + r'(?:(".+?")|(?:[\[`])?([a-z0-9_]+)(?:[\]`])?) ' + r"+[a-z0-9_ ]+? +UNIQUE" ) for match in re.finditer(UNIQUE_PATTERN, table_data, re.I): diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index 2e0eccc96b..6230c7f945 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -42,6 +42,7 @@ from sqlalchemy.dialects.sqlite import pysqlite as pysqlite_dialect from sqlalchemy.engine.url import make_url from sqlalchemy.schema import CreateTable from sqlalchemy.schema import FetchedValue +from sqlalchemy.sql.elements import quoted_name from sqlalchemy.testing import assert_raises from sqlalchemy.testing import assert_raises_message from sqlalchemy.testing import AssertsCompiledSQL @@ -2401,6 +2402,55 @@ class ConstraintReflectionTest(fixtures.TestBase): ], ) + @testing.combinations( + ("plain_name", "plain_name"), + ("name with spaces", "name with spaces"), + ("plainname", "plainname"), + ("[Code]", "[Code]"), + (quoted_name("[Code]", quote=False), "Code"), + argnames="colname,expected", + ) + @testing.combinations( + "uq", "uq_inline", "pk", "ix", argnames="constraint_type" + ) + def test_constraint_cols( + self, colname, expected, constraint_type, connection, metadata + ): + if constraint_type == "uq_inline": + t = Table("t", metadata, Column(colname, Integer)) + connection.exec_driver_sql( + """ + CREATE TABLE t (%s INTEGER UNIQUE) + """ + % connection.dialect.identifier_preparer.quote(colname) + ) + else: + t = Table("t", metadata, Column(colname, Integer)) + if constraint_type == "uq": + constraint = UniqueConstraint(t.c[colname]) + elif constraint_type == "pk": + constraint = PrimaryKeyConstraint(t.c[colname]) + elif constraint_type == "ix": + constraint = Index("some_index", t.c[colname]) + else: + assert False + + t.append_constraint(constraint) + + t.create(connection) + + if constraint_type in ("uq", "uq_inline"): + const = inspect(connection).get_unique_constraints("t")[0] + eq_(const["column_names"], [expected]) + elif constraint_type == "pk": + const = inspect(connection).get_pk_constraint("t") + eq_(const["constrained_columns"], [expected]) + elif constraint_type == "ix": + const = inspect(connection).get_indexes("t")[0] + eq_(const["column_names"], [expected]) + else: + assert False + class SavepointTest(fixtures.TablesTest):