may now be a mix of lists and tuples. (Previously members
were always lists.)
-
+- schema
+ - Added "sorted_tables" accessor to MetaData, which returns
+ Table objects sorted in order of dependency as a list.
+ This deprecates the MetaData.table_iterator() method.
+ The "reverse=False" keyword argument has also been
+ removed from util.sort_tables(); use the Python
+ 'reversed' function to reverse the results.
+ [ticket:1033]
+
- sql
- Temporarily rolled back the "ORDER BY" enhancement from
[ticket:1068]. This feature is on hold pending further
for t in mapper.tables:
table_to_mapper[t] = mapper
- for table in sqlutil.sort_tables(table_to_mapper.keys(), reverse=True):
+ for table in reversed(sqlutil.sort_tables(table_to_mapper.keys())):
delete = {}
for state, mapper, connection in tups:
if table not in mapper._pks_by_table:
# TODO: scan all other tables and remove FK _column
del self.tables[table.key]
+ @util.deprecated('Deprecated. Use ``metadata.sorted_tables``')
def table_iterator(self, reverse=True, tables=None):
from sqlalchemy.sql.util import sort_tables
if tables is None:
tables = self.tables.values()
else:
tables = set(tables).intersection(self.tables.values())
- return iter(sort_tables(tables, reverse=reverse))
-
+ ret = sort_tables(tables)
+ if reverse:
+ ret = reversed(ret)
+ return iter(ret)
+
+ @property
+ def sorted_tables(self):
+ from sqlalchemy.sql.util import sort_tables
+ return sort_tables(self.tables.values())
+
def reflect(self, bind=None, schema=None, only=None):
"""Load all available table definitions from the database.
checkfirst
Defaults to True, don't issue CREATEs for tables already present
in the target database.
+
"""
-
if bind is None:
bind = _bind_or_error(self)
for listener in self.ddl_listeners['before-create']:
import string, re
from sqlalchemy import schema, engine, util, exc
-from sqlalchemy.sql import operators, functions
+from sqlalchemy.sql import operators, functions, util as sql_util
from sqlalchemy.sql import expression as sql
RESERVED_WORDS = set([
return not self.checkfirst or not self.dialect.has_table(self.connection, table.name, schema=table.schema)
def visit_metadata(self, metadata):
- collection = [t for t in metadata.table_iterator(reverse=False, tables=self.tables) if self._can_create(t)]
+ if self.tables:
+ tables = self.tables
+ else:
+ tables = metadata.tables.values()
+ collection = [t for t in sql_util.sort_tables(tables) if self._can_create(t)]
for table in collection:
self.traverse_single(table)
if self.dialect.supports_alter:
self.dialect = dialect
def visit_metadata(self, metadata):
- collection = [t for t in metadata.table_iterator(reverse=True, tables=self.tables) if self._can_drop(t)]
+ if self.tables:
+ tables = self.tables
+ else:
+ tables = metadata.tables.values()
+ collection = [t for t in reversed(sql_util.sort_tables(tables)) if self._can_drop(t)]
if self.dialect.supports_alter:
for alterable in self.find_alterables(collection):
self.drop_foreignkey(alterable)
"""Utility functions that build upon SQL and Schema constructs."""
-def sort_tables(tables, reverse=False):
+def sort_tables(tables):
"""sort a collection of Table objects in order of their foreign-key dependency."""
tuples = []
for table in tables:
visitors.traverse(table, {'schema_visitor':True}, {'foreign_key':visit_foreign_key})
- sequence = topological.sort(tuples, tables)
- if reverse:
- return reversed(sequence)
- else:
- return sequence
+ return topological.sort(tuples, tables)
def search(clause, target):
if not clause:
def tearDown(self):
# no need to run deletes if tables are recreated on setup
if self.run_define_tables != 'each' and self.run_deletes:
- for table in self.metadata.table_iterator(reverse=True):
+ for table in reversed(self.metadata.sorted_tables):
try:
table.delete().execute().close()
except sa.exc.DBAPIError, ex:
table = self.tables[table]
headers[table] = data[0]
rows[table] = data[1:]
- for table in self.metadata.table_iterator(reverse=False):
+ for table in self.metadata.sorted_tables:
if table not in headers:
continue
table.bind.execute(
)
def test_sorter( self ):
- tables = metadata.table_iterator(reverse=False)
+ tables = metadata.sorted_tables
table_names = [t.name for t in tables]
self.assert_( table_names == ['users', 'orders', 'items', 'email_addresses'] or table_names == ['users', 'email_addresses', 'orders', 'items'])
Column('col2', sa.Integer, sa.ForeignKey('someschema.table1.col1')),
schema='someschema')
# ensure this doesnt crash
- print [t for t in metadata.table_iterator()]
+ print [t for t in metadata.sorted_tables]
buf = StringIO.StringIO()
def foo(s, p=None):
buf.write(s)
u1 = User(name='u1', a='a', b='b')
eq_(u1.a, 'a')
- eq_(User.a.get_history(u1), (['a'], [], []))
+ eq_(User.a.get_history(u1), (['a'], (), ()))
sess = create_session()
sess.save(u1)
sess.flush()
Base = decl.declarative_base(testing.db)
def tearDown(self):
- for t in reflection_metadata.table_iterator():
+ for t in reflection_metadata.sorted_tables:
t.delete().execute()
def tearDownAll(self):
# no need to run deletes if tables are recreated on setup
if self.run_define_tables != 'each' and self.run_deletes:
- for table in self.metadata.table_iterator(reverse=True):
+ for table in reversed(self.metadata.sorted_tables):
try:
table.delete().execute().close()
except sa.exc.DBAPIError, ex:
table = self.tables[table]
headers[table] = data[0]
rows[table] = data[1:]
- for table in self.metadata.table_iterator(reverse=False):
+ for table in self.metadata.sorted_tables:
if table not in headers:
continue
table.bind.execute(
def _load_fixtures():
- for table in fixture_metadata.table_iterator(reverse=False):
+ for table in fixture_metadata.sorted_tables:
table.info[('fixture', 'loader')]()
def run_inserts_for(table, bind=None):
for j1 in (None, _b_table.c.a_id==_a_table.c.id, _a_table.c.id==_b_table.c.a_id):
for j2 in (None, _b_table.c.a_id==_c_table.c.b_a_id, _c_table.c.b_a_id==_b_table.c.a_id):
self._do_test(j1, j2)
- for t in _a_table.metadata.table_iterator(reverse=True):
+ for t in reversed(_a_table.metadata.sorted_tables):
t.delete().execute().close()
def _do_test(self, j1, j2):
metadata.drop_all()
def tearDown(self):
clear_mappers()
- for t in metadata.table_iterator(reverse=True):
+ for t in reversed(metadata.sorted_tables):
t.delete().execute()
def testjointo(self):
if not self.keep_mappers:
clear_mappers()
if not self.keep_data:
- for t in _otest_metadata.table_iterator(reverse=True):
+ for t in reversed(_otest_metadata.sorted_tables):
try:
t.delete().execute().close()
except Exception, e:
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
