Take schema name into account when querying sqlite_master

Fixed bug where SQLite CHECK constraint reflection would fail
if the referenced table were in a remote schema, e.g. on SQLite a
remote database referred to by ATTACH.

Also add suite support for general CHECK constraint reflection.

Change-Id: I073a72cb47dc4f8c5683000d708768523759332f
Fixes: #4099
(cherry picked from commit 68b52c48b775f9a99d0bc3666ebe02c54e401303)

diff --git a/doc/build/changelog/unreleased_11/4099.rst b/doc/build/changelog/unreleased_11/4099.rst
new file mode 100644 (file)
index 0000000..e41e900
--- /dev/null
+++ b/doc/build/changelog/unreleased_11/4099.rst
@@ -0,0 +1,8 @@
+.. change:
+    :tags: bug, sqlite
+    :tickets: 4099
+    :versions: 1.2.0b3
+
+    Fixed bug where SQLite CHECK constraint reflection would fail
+    if the referenced table were in a remote schema, e.g. on SQLite a
+    remote database referred to by ATTACH.
\ No newline at end of file

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 4bc60be62e529f29e13a6ce7a03c42f4c6e35923..d8ce7f394caa46f9ec38e1bf044b375eae5672f7 100644 (file)
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1545,14 +1545,19 @@ class SQLiteDialect(default.DefaultDialect):
     def _get_table_sql(self, connection, table_name, schema=None, **kw):
         try:
             s = ("SELECT sql FROM "
-                 " (SELECT * FROM sqlite_master UNION ALL "
-                 "  SELECT * FROM sqlite_temp_master) "
-                 "WHERE name = '%s' "
-                 "AND type = 'table'") % table_name
+                 " (SELECT * FROM %(schema)ssqlite_master UNION ALL "
+                 "  SELECT * FROM %(schema)ssqlite_temp_master) "
+                 "WHERE name = '%(table)s' "
+                 "AND type = 'table'" % {
+                     "schema": ("%s." % schema) if schema else "",
+                     "table": table_name})
             rs = connection.execute(s)
         except exc.DBAPIError:
-            s = ("SELECT sql FROM sqlite_master WHERE name = '%s' "
-                 "AND type = 'table'") % table_name
+            s = ("SELECT sql FROM %(schema)ssqlite_master "
+                 "WHERE name = '%(table)s' "
+                 "AND type = 'table'" % {
+                     "schema": ("%s." % schema) if schema else "",
+                     "table": table_name})
             rs = connection.execute(s)
         return rs.scalar()
 

diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index e4491ca7c5fbb28fdb927a6f75dc70d6c2cadb92..e88dc9417e1792e0097acbf07877c0dc4ed01c32 100644 (file)
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -381,6 +381,11 @@ class SuiteRequirements(Requirements):
         """target dialect supports reflection of unique constraints"""
         return exclusions.open()
 
+    @property
+    def check_constraint_reflection(self):
+        """target dialect supports reflection of check constraints"""
+        return exclusions.closed()
+
     @property
     def duplicate_key_raises_integrity_error(self):
         """target dialect raises IntegrityError when reporting an INSERT

diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index ed6a33b6d44b0fe8cab0bcd3a458d221e99f146d..f2e178bd36444dcea5d376680491548ea1d32b1f 100644 (file)
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -16,6 +16,7 @@ from sqlalchemy.schema import DDL, Index
 from sqlalchemy import event
 from sqlalchemy.sql.elements import quoted_name
 from sqlalchemy import ForeignKey
+import re
 
 metadata, users = None, None
 
@@ -640,6 +641,50 @@ class ComponentReflectionTest(fixtures.TablesTest):
             refl.pop('duplicates_index', None)
             eq_(orig, refl)
 
+    @testing.requires.check_constraint_reflection
+    def test_get_check_constraints(self):
+        self._test_get_check_constraints()
+
+    @testing.requires.check_constraint_reflection
+    @testing.requires.schemas
+    def test_get_check_constraints_schema(self):
+        self._test_get_check_constraints(schema=testing.config.test_schema)
+
+    @testing.provide_metadata
+    def _test_get_check_constraints(self, schema=None):
+        orig_meta = self.metadata
+        Table(
+            'sa_cc', orig_meta,
+            Column('a', Integer()),
+            sa.CheckConstraint('a > 1 AND a < 5', name='cc1'),
+            sa.CheckConstraint('a = 1 OR (a > 2 AND a < 5)', name='cc2'),
+            schema=schema
+        )
+
+        orig_meta.create_all()
+
+        inspector = inspect(orig_meta.bind)
+        reflected = sorted(
+            inspector.get_check_constraints('sa_cc', schema=schema),
+            key=operator.itemgetter('name')
+        )
+
+        reflected = [
+            {"name": item["name"],
+             # trying to minimize effect of quoting, parenthesis, etc.
+             # may need to add more to this as new dialects get CHECK
+             # constraint reflection support
+             "sqltext": re.sub(r"[`'\(\)]", '', item["sqltext"].lower())}
+            for item in reflected
+        ]
+        eq_(
+            reflected,
+            [
+                {'name': 'cc1', 'sqltext': 'a > 1 and a < 5'},
+                {'name': 'cc2', 'sqltext': 'a = 1 or a > 2 and a < 5'}
+            ]
+        )
+
     @testing.provide_metadata
     def _test_get_view_definition(self, schema=None):
         meta = self.metadata

diff --git a/test/requirements.py b/test/requirements.py
index 52fe37f9f933d5311f37b84d4c65d9e4e2abbdf0..3c80b0ad30c09900224d06d8d84b7c95abd53d65 100644 (file)
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -328,6 +328,7 @@ class DefaultRequirements(SuiteRequirements):
     def unique_constraint_reflection_no_index_overlap(self):
         return self.unique_constraint_reflection + skip_if("mysql")
 
+
     @property
     def check_constraint_reflection(self):
         return fails_on_everything_except(