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 (file)
index 0000000..828dd54
--- /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 236f6eaeff85f72a142485e02171642e3b00b2cd..385de373ead61e0c7cac7d6ef1ff74b307b54717 100644 (file)
--- 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 d7021a3432f04af814d52c3d7f6aef5b3d91425f..9658fec8329f7a8aada70848ada8989a226b1fb0 100644 (file)
--- 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):