series as well. For changes that are specific to 1.0 with an emphasis
on compatibility concerns, see :doc:`/changelog/migration_10`.
+ .. change::
+ :tags: bug, sqlite
+ :tickets: 3204
+
+ Added :meth:`.Inspector.get_temp_table_names` and
+ :meth:`.Inspector.get_temp_view_names`; currently, only the
+ SQLite dialect supports these methods. The return of temporary
+ table and view names has been **removed** from SQLite's version
+ of :meth:`.Inspector.get_table_names` and
+ :meth:`.Inspector.get_view_names`; other database backends cannot
+ support this information (such as MySQL), and the scope of operation
+ is different in that the tables can be local to a session and
+ typically aren't supported in remote schemas.
+
+ .. seealso::
+
+ :ref:`change_3204`
+
.. change::
:tags: feature, postgresql
:tickets: 2891
:ticket:`3182`
+.. _change_3204:
+
+SQLite has distinct methods for temporary table/view name reporting
+-------------------------------------------------------------------
+
+The :meth:`.Inspector.get_table_names` and :meth:`.Inspector.get_view_names`
+methods in the case of SQLite would also return the names of temporary
+tables and views, which is not provided by any other dialect (in the case
+of MySQL at least it is not even possible). This logic has been moved
+out to two new methods :meth:`.Inspector.get_temp_table_names` and
+:meth:`.Inspector.get_temp_view_names`.
+
+Note that reflection of a specific named temporary table or temporary view,
+either by ``Table('name', autoload=True)`` or via methods like
+:meth:`.Inspector.get_columns` continues to function for most if not all
+dialects. For SQLite specifically, there is a bug fix for UNIQUE constraint
+reflection from temp tables as well, which is :ticket:`3203`.
+
+:ticket:`3204`
+
.. _change_2984:
Drizzle Dialect is now an External Dialect
if schema is not None:
qschema = self.identifier_preparer.quote_identifier(schema)
master = '%s.sqlite_master' % qschema
- s = ("SELECT name FROM %s "
- "WHERE type='table' ORDER BY name") % (master,)
- rs = connection.execute(s)
else:
- try:
- s = ("SELECT name FROM "
- " (SELECT * FROM sqlite_master UNION ALL "
- " SELECT * FROM sqlite_temp_master) "
- "WHERE type='table' ORDER BY name")
- rs = connection.execute(s)
- except exc.DBAPIError:
- s = ("SELECT name FROM sqlite_master "
- "WHERE type='table' ORDER BY name")
- rs = connection.execute(s)
+ master = "sqlite_master"
+ s = ("SELECT name FROM %s "
+ "WHERE type='table' ORDER BY name") % (master,)
+ rs = connection.execute(s)
+ return [row[0] for row in rs]
+
+ @reflection.cache
+ def get_temp_table_names(self, connection, **kw):
+ s = "SELECT name FROM sqlite_temp_master "\
+ "WHERE type='table' ORDER BY name "
+ rs = connection.execute(s)
+
+ return [row[0] for row in rs]
+
+ @reflection.cache
+ def get_temp_view_names(self, connection, **kw):
+ s = "SELECT name FROM sqlite_temp_master "\
+ "WHERE type='view' ORDER BY name "
+ rs = connection.execute(s)
return [row[0] for row in rs]
if schema is not None:
qschema = self.identifier_preparer.quote_identifier(schema)
master = '%s.sqlite_master' % qschema
- s = ("SELECT name FROM %s "
- "WHERE type='view' ORDER BY name") % (master,)
- rs = connection.execute(s)
else:
- try:
- s = ("SELECT name FROM "
- " (SELECT * FROM sqlite_master UNION ALL "
- " SELECT * FROM sqlite_temp_master) "
- "WHERE type='view' ORDER BY name")
- rs = connection.execute(s)
- except exc.DBAPIError:
- s = ("SELECT name FROM sqlite_master "
- "WHERE type='view' ORDER BY name")
- rs = connection.execute(s)
+ master = "sqlite_master"
+ s = ("SELECT name FROM %s "
+ "WHERE type='view' ORDER BY name") % (master,)
+ rs = connection.execute(s)
return [row[0] for row in rs]
def get_table_names(self, connection, schema=None, **kw):
"""Return a list of table names for `schema`."""
- raise NotImplementedError
+ raise NotImplementedError()
+
+ def get_temp_table_names(self, connection, schema=None, **kw):
+ """Return a list of temporary table names on the given connection,
+ if supported by the underlying backend.
+
+ """
+
+ raise NotImplementedError()
def get_view_names(self, connection, schema=None, **kw):
"""Return a list of all view names available in the database.
raise NotImplementedError()
+ def get_temp_view_names(self, connection, schema=None, **kw):
+ """Return a list of temporary view names on the given connection,
+ if supported by the underlying backend.
+
+ """
+
+ raise NotImplementedError()
+
def get_view_definition(self, connection, view_name, schema=None, **kw):
"""Return view definition.
tnames = list(topological.sort(tuples, tnames))
return tnames
+ def get_temp_table_names(self):
+ """return a list of temporary table names for the current bind.
+
+ This method is unsupported by most dialects; currently
+ only SQLite implements it.
+
+ .. versionadded:: 1.0.0
+
+ """
+ return self.dialect.get_temp_table_names(
+ self.bind, info_cache=self.info_cache)
+
+ def get_temp_view_names(self):
+ """return a list of temporary view names for the current bind.
+
+ This method is unsupported by most dialects; currently
+ only SQLite implements it.
+
+ .. versionadded:: 1.0.0
+
+ """
+ return self.dialect.get_temp_view_names(
+ self.bind, info_cache=self.info_cache)
+
def get_table_options(self, table_name, schema=None, **kw):
"""Return a dictionary of options specified when the table of the
given name was created.
def foreign_key_constraint_reflection(self):
return exclusions.open()
+ @property
+ def temp_table_reflection(self):
+ return exclusions.open()
+
+ @property
+ def temp_table_names(self):
+ """target dialect supports listing of temporary table names"""
+ return exclusions.closed()
+
+ @property
+ def temporary_views(self):
+ """target database supports temporary views"""
+ return exclusions.closed()
+
@property
def index_reflection(self):
return exclusions.open()
cls.define_index(metadata, users)
if testing.requires.view_column_reflection.enabled:
cls.define_views(metadata, schema)
+ if not schema and testing.requires.temp_table_reflection.enabled:
+ cls.define_temp_tables(metadata)
+
+ @classmethod
+ def define_temp_tables(cls, metadata):
+ temp_table = Table(
+ "user_tmp", metadata,
+ Column("id", sa.INT, primary_key=True),
+ Column('name', sa.VARCHAR(50)),
+ Column('foo', sa.INT),
+ sa.UniqueConstraint('name', name='user_tmp_uq'),
+ sa.Index("user_tmp_ix", "foo"),
+ prefixes=['TEMPORARY']
+ )
+ if testing.requires.view_reflection.enabled and \
+ testing.requires.temporary_views.enabled:
+ event.listen(
+ temp_table, "after_create",
+ DDL("create temporary view user_tmp_v as "
+ "select * from user_tmp")
+ )
@classmethod
def define_index(cls, metadata, users):
users, addresses, dingalings = self.tables.users, \
self.tables.email_addresses, self.tables.dingalings
insp = inspect(meta.bind)
+
if table_type == 'view':
table_names = insp.get_view_names(schema)
table_names.sort()
answer = ['dingalings', 'email_addresses', 'users']
eq_(sorted(table_names), answer)
+ @testing.requires.temp_table_names
+ def test_get_temp_table_names(self):
+ insp = inspect(self.metadata.bind)
+ temp_table_names = insp.get_temp_table_names()
+ eq_(sorted(temp_table_names), ['user_tmp'])
+
+ @testing.requires.view_reflection
+ @testing.requires.temp_table_names
+ @testing.requires.temporary_views
+ def test_get_temp_view_names(self):
+ insp = inspect(self.metadata.bind)
+ temp_table_names = insp.get_temp_view_names()
+ eq_(sorted(temp_table_names), ['user_tmp_v'])
+
@testing.requires.table_reflection
def test_get_table_names(self):
self._test_get_table_names()
def test_get_columns_with_schema(self):
self._test_get_columns(schema=testing.config.test_schema)
+ @testing.requires.temp_table_reflection
+ def test_get_temp_table_columns(self):
+ meta = MetaData(testing.db)
+ user_tmp = self.tables.user_tmp
+ insp = inspect(meta.bind)
+ cols = insp.get_columns('user_tmp')
+ self.assert_(len(cols) > 0, len(cols))
+
+ for i, col in enumerate(user_tmp.columns):
+ eq_(col.name, cols[i]['name'])
+
+ @testing.requires.temp_table_reflection
+ @testing.requires.view_column_reflection
+ @testing.requires.temporary_views
+ def test_get_temp_view_columns(self):
+ insp = inspect(self.metadata.bind)
+ cols = insp.get_columns('user_tmp_v')
+ eq_(
+ [col['name'] for col in cols],
+ ['id', 'name', 'foo']
+ )
+
@testing.requires.view_column_reflection
def test_get_view_columns(self):
self._test_get_columns(table_type='view')
def test_get_unique_constraints(self):
self._test_get_unique_constraints()
+ @testing.requires.temp_table_reflection
+ def test_get_temp_table_unique_constraints(self):
+ insp = inspect(self.metadata.bind)
+ eq_(
+ insp.get_unique_constraints('user_tmp'),
+ [{'column_names': ['name'], 'name': 'user_tmp_uq'}]
+ )
+
+ @testing.requires.temp_table_reflection
+ def test_get_temp_table_indexes(self):
+ insp = inspect(self.metadata.bind)
+ indexes = insp.get_indexes('user_tmp')
+ eq_(
+ # TODO: we need to add better filtering for indexes/uq constraints
+ # that are doubled up
+ [idx for idx in indexes if idx['name'] == 'user_tmp_ix'],
+ [{'unique': False, 'column_names': ['foo'], 'name': 'user_tmp_ix'}]
+ )
+
+
@testing.requires.unique_constraint_reflection
@testing.requires.schemas
def test_get_unique_constraints_with_schema(self):
finally:
cx.execute('DETACH DATABASE test_schema')
- @testing.exclude('sqlite', '<', (2, 6), 'no database support')
- def test_temp_table_reflection(self):
- cx = testing.db.connect()
- try:
- cx.execute('CREATE TEMPORARY TABLE tempy (id INT)')
- assert 'tempy' in cx.dialect.get_table_names(cx, None)
- meta = MetaData(cx)
- tempy = Table('tempy', meta, autoload=True)
- assert len(tempy.c) == 1
- meta.drop_all()
- except:
- try:
- cx.execute('DROP TABLE tempy')
- except exc.DBAPIError:
- pass
- raise
-
def test_file_path_is_absolute(self):
d = pysqlite_dialect.dialect()
eq_(
"sqlite"
)
+ @property
+ def temp_table_names(self):
+ """target dialect supports listing of temporary table names"""
+
+ return only_on(['sqlite'])
+
+ @property
+ def temporary_views(self):
+ """target database supports temporary views"""
+ return only_on(['sqlite', 'postgresql'])
+
@property
def update_nowait(self):
"""Target database must support SELECT...FOR UPDATE NOWAIT"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
