where specific capabilities and features can
be enabled or disabled for testing.
+ - [bug] The Inspector.get_table_names()
+ order_by="foreign_key" feature now sorts
+ tables by dependee first, to be consistent
+ with util.sort_tables and metadata.sorted_tables.
+
- [bug] Fixed bug whereby if a database restart
affected multiple connections, each
connection would individually invoke a new
return []
def get_table_names(self, schema=None, order_by=None):
- """Return all table names in `schema`.
+ """Return all table names in referred to within a particular schema.
+
+ The names are expected to be real tables only, not views.
+ Views are instead returned using the :meth:`.get_view_names`
+ method.
+
+
+ :param schema: Schema name. If ``schema`` is left at ``None``, the
+ database's default schema is
+ used, else the named schema is searched. If the database does not
+ support named schemas, behavior is undefined if ``schema`` is not
+ passed as ``None``.
- :param schema: Optional, retrieve names from a non-default schema.
:param order_by: Optional, may be the string "foreign_key" to sort
- the result on foreign key dependencies.
+ the result on foreign key dependencies.
+
+ .. versionchanged:: 0.8 the "foreign_key" sorting sorts tables
+ in order of dependee to dependent; that is, in creation
+ order, rather than in drop order. This is to maintain
+ consistency with similar features such as
+ :attr:`.MetaData.sorted_tables` and :func:`.util.sort_tables`.
+
+ .. seealso::
+
+ :attr:`.MetaData.sorted_tables`
- This should probably not return view names or maybe it should return
- them with an indicator t or v.
"""
if hasattr(self.dialect, 'get_table_names'):
else:
tnames = self.engine.table_names(schema)
if order_by == 'foreign_key':
- import random
- random.shuffle(tnames)
-
tuples = []
for tname in tnames:
for fkey in self.get_foreign_keys(tname, schema):
if tname != fkey['referred_table']:
- tuples.append((tname, fkey['referred_table']))
+ tuples.append((fkey['referred_table'], tname))
tnames = list(topological.sort(tuples, tnames))
return tnames
@property
def sorted_tables(self):
- """Returns a list of ``Table`` objects sorted in order of
- dependency.
+ """Returns a list of :class:`.Table` objects sorted in order of
+ foreign key dependency.
+
+ The sorting will place :class:`.Table` objects that have dependencies
+ first, before the dependencies themselves, representing the
+ order in which they can be created. To get the order in which
+ the tables would be dropped, use the ``reversed()`` Python built-in.
+
+ .. seealso::
+
+ :meth:`.Inspector.sorted_tables`
+
"""
return sqlutil.sort_tables(self.tables.itervalues())
for column in element.columns
]
+
+class _DropView(_CreateDropBase):
+ """Semi-public 'DROP VIEW' construct.
+
+ Used by the test suite for dialect-agnostic drops of views.
+ This object will eventually be part of a public "view" API.
+
+ """
+ __visit_name__ = "drop_view"
+
class CreateColumn(visitors.Visitable):
"""Represent a :class:`.Column` as rendered in a CREATE TABLE statement,
via the :class:`.CreateTable` construct.
def visit_drop_table(self, drop):
return "\nDROP TABLE " + self.preparer.format_table(drop.element)
+ def visit_drop_view(self, drop):
+ return "\nDROP VIEW " + self.preparer.format_table(drop.element)
+
def _index_identifier(self, ident):
if isinstance(ident, sql._truncated_label):
max = self.dialect.max_index_name_length or \
"""Utility functions that build upon SQL and Schema constructs."""
def sort_tables(tables, skip_fn=None):
- """sort a collection of Table objects in order of their foreign-key dependency."""
+ """sort a collection of Table objects in order of
+ their foreign-key dependency."""
tables = list(tables)
tuples = []
for table in tables:
visitors.traverse(table,
- {'schema_visitor':True},
- {'foreign_key':visit_foreign_key})
+ {'schema_visitor': True},
+ {'foreign_key': visit_foreign_key})
tuples.extend(
[parent, table] for parent in table._extra_dependencies
@post
def _prep_testing_database(options, file_config):
from sqlalchemy.testing import engines
- from sqlalchemy import schema
+ from sqlalchemy import schema, inspect
# also create alt schemas etc. here?
if options.dropfirst:
e = engines.utf8_engine()
- existing = e.table_names()
- if existing:
- print "Dropping existing tables in database: " + db_url
- try:
- print "Tables: %s" % ', '.join(existing)
- except:
- pass
- print "Abort within 5 seconds..."
- time.sleep(5)
- md = schema.MetaData(e, reflect=True)
- md.drop_all()
+ inspector = inspect(e)
+
+ for vname in inspector.get_view_names():
+ e.execute(schema._DropView(schema.Table(vname, schema.MetaData())))
+
+ for vname in inspector.get_view_names(schema="test_schema"):
+ e.execute(schema._DropView(
+ schema.Table(vname,
+ schema.MetaData(), schema="test_schema")))
+
+ for tname in reversed(inspector.get_table_names(order_by="foreign_key")):
+ e.execute(schema.DropTable(schema.Table(tname, schema.MetaData())))
+
+ for tname in reversed(inspector.get_table_names(
+ order_by="foreign_key", schema="test_schema")):
+ e.execute(schema.DropTable(
+ schema.Table(tname, schema.MetaData(), schema="test_schema")))
+
e.dispose()
)
cls.define_index(metadata, users)
- cls.define_views(metadata, schema=None)
+ cls.define_views(metadata, schema)
@classmethod
def define_index(cls, metadata, users):
view_name = fullname + '_v'
query = "CREATE VIEW %s AS SELECT * FROM %s" % (
view_name, fullname)
+
event.listen(
metadata,
"after_create",
table_names = insp.get_table_names(schema,
order_by=order_by)
if order_by == 'foreign_key':
- answer = ['dingalings', 'email_addresses', 'users']
+ answer = ['users', 'email_addresses', 'dingalings']
eq_(table_names, answer)
else:
answer = ['dingalings', 'email_addresses', 'users']
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
