]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
improve reflection of inline UNIQUE constraints
authorMike Bayer <mike_mp@zzzcomputing.com>
Sun, 20 Feb 2022 14:51:22 +0000 (09:51 -0500)
committerMike Bayer <mike_mp@zzzcomputing.com>
Mon, 21 Feb 2022 01:39:59 +0000 (20:39 -0500)
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)

doc/build/changelog/unreleased_14/7736.rst [new file with mode: 0644]
lib/sqlalchemy/dialects/sqlite/base.py
test/dialect/test_sqlite.py

diff --git a/doc/build/changelog/unreleased_14/7736.rst b/doc/build/changelog/unreleased_14/7736.rst
new file mode 100644 (file)
index 0000000..828dd54
--- /dev/null
@@ -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.
index bcea17620f3c99c298a88ee0a0f2628dcce5ea9d..7ba9700d709d85d9fbf76932d9a305e0e0f30a58 100644 (file)
@@ -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):
index 2e0eccc96bb403199c61dac4f8cd5b88b360d25e..6230c7f94595f754ad0cb2f5a114048e81fa4d20 100644 (file)
@@ -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):