From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Sun, 20 Feb 2022 14:51:22 +0000 (-0500)
Subject: improve reflection of inline UNIQUE constraints
X-Git-Tag: rel_2_0_0b1~477^2
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=834af17a469fd1893acf20225e8400c0c908053f;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
---

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 236f6eaeff..385de373ea 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -2395,7 +2395,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 d7021a3432..9658fec832 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -41,6 +41,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
@@ -2387,6 +2388,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):