diff options
author | Johannes Erdfelt <johannes@erdfelt.com> | 2014-09-17 07:52:34 -0700 |
---|---|---|
committer | Johannes Erdfelt <johannes@erdfelt.com> | 2014-09-17 08:01:01 -0700 |
commit | 9d402e204d77da680472cbfb9813e437eb187944 (patch) | |
tree | 7be9fc6007ac44758ccfad05a1e64c046bb74a0f | |
parent | a985f84ed6223e7a7348dd6126f8de92012b635f (diff) | |
download | sqlalchemy-9d402e204d77da680472cbfb9813e437eb187944.tar.gz |
Handle sqlite get_unique_constraints() call for temporary tables
The sqlite get_unique_constraints() implementation did not do a union
against the sqlite_temp_master table like other code does. This could
result in an exception being raised if get_unique_constraints() was
called against a temporary table.
-rw-r--r-- | lib/sqlalchemy/dialects/sqlite/base.py | 28 | ||||
-rw-r--r-- | test/dialect/test_sqlite.py | 18 |
2 files changed, 36 insertions, 10 deletions
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py index af793d275..c76ef6afd 100644 --- a/lib/sqlalchemy/dialects/sqlite/base.py +++ b/lib/sqlalchemy/dialects/sqlite/base.py @@ -1097,16 +1097,24 @@ class SQLiteDialect(default.DefaultDialect): @reflection.cache def get_unique_constraints(self, connection, table_name, schema=None, **kw): - UNIQUE_SQL = """ - SELECT sql - FROM - sqlite_master - WHERE - type='table' AND - name=:table_name - """ - c = connection.execute(UNIQUE_SQL, table_name=table_name) - table_data = c.fetchone()[0] + try: + s = ("SELECT sql FROM " + " (SELECT * FROM sqlite_master UNION ALL " + " SELECT * FROM sqlite_temp_master) " + "WHERE name = '%s' " + "AND type = 'table'") % table_name + rs = connection.execute(s) + except exc.DBAPIError: + s = ("SELECT sql FROM sqlite_master WHERE name = '%s' " + "AND type = 'table'") % table_name + rs = connection.execute(s) + row = rs.fetchone() + if row is None: + # sqlite won't return the schema for the sqlite_master or + # sqlite_temp_master tables from this query. These tables + # don't have any unique constraints anyway. + return [] + table_data = row[0] UNIQUE_PATTERN = 'CONSTRAINT (\w+) UNIQUE \(([^\)]+)\)' return [ diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py index e77a03980..6fc644689 100644 --- a/test/dialect/test_sqlite.py +++ b/test/dialect/test_sqlite.py @@ -575,6 +575,24 @@ class DialectTest(fixtures.TestBase, AssertsExecutionResults): finally: meta.drop_all() + def test_get_unique_constraints(self): + meta = MetaData(testing.db) + t1 = Table('foo', meta, Column('f', Integer), + UniqueConstraint('f', name='foo_f')) + t2 = Table('bar', meta, Column('b', Integer), + UniqueConstraint('b', name='bar_b'), + prefixes=['TEMPORARY']) + meta.create_all() + from sqlalchemy.engine.reflection import Inspector + try: + inspector = Inspector(testing.db) + eq_(inspector.get_unique_constraints('foo'), + [{'column_names': [u'f'], 'name': u'foo_f'}]) + eq_(inspector.get_unique_constraints('bar'), + [{'column_names': [u'b'], 'name': u'bar_b'}]) + finally: + meta.drop_all() + class SQLTest(fixtures.TestBase, AssertsCompiledSQL): |