- restate sort_tables in terms of a more fine grained

sort_tables_and_constraints function.
- The DDL generation system of :meth:`.MetaData.create_all`
and :meth:`.Metadata.drop_all` has been enhanced to in most
cases automatically handle the case of mutually dependent
foreign key constraints; the need for the
:paramref:`.ForeignKeyConstraint.use_alter` flag is greatly
reduced.  The system also works for constraints which aren't given
a name up front; only in the case of DROP is a name required for
at least one of the constraints involved in the cycle.
fixes #3282

10 files changed, 441 insertions, 75 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index c868f58b2..5f990ea4e 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1767,10 +1767,10 @@ class MySQLCompiler(compiler.SQLCompiler):
 #       creation of foreign key constraints fails."
 
 class MySQLDDLCompiler(compiler.DDLCompiler):
-    def create_table_constraints(self, table):
+    def create_table_constraints(self, table, **kw):
         """Get table constraints."""
         constraint_string = super(
-            MySQLDDLCompiler, self).create_table_constraints(table)
+            MySQLDDLCompiler, self).create_table_constraints(table, **kw)
 
         # why self.dialect.name and not 'mysql'?  because of drizzle
         is_innodb = 'engine' in table.dialect_options[self.dialect.name] and \
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index e0b2875e8..3d7b0788b 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -201,6 +201,15 @@ new connections through the usage of events::
         cursor.execute("PRAGMA foreign_keys=ON")
         cursor.close()
 
+.. warning::
+
+    When SQLite foreign keys are enabled, it is **not possible**
+    to emit CREATE or DROP statements for tables that contain
+    mutually-dependent foreign key constraints;
+    to emit the DDL for these tables requires that ALTER TABLE be used to
+    create or drop these constraints separately, for which SQLite has
+    no support.
+
 .. seealso::
 
     `SQLite Foreign Key Support <http://www.sqlite.org/foreignkeys.html>`_
@@ -208,6 +217,9 @@ new connections through the usage of events::
 
     :ref:`event_toplevel` - SQLAlchemy event API.
 
+    :ref:`use_alter` - more information on SQLAlchemy's facilities for handling
+     mutually-dependent foreign key constraints.
+
 .. _sqlite_type_reflection:
 
 Type Reflection
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 25f084c15..6e102aad6 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -173,7 +173,14 @@ class Inspector(object):
          passed as ``None``.  For special quoting, use :class:`.quoted_name`.
 
         :param order_by: Optional, may be the string "foreign_key" to sort
-         the result on foreign key dependencies.
+         the result on foreign key dependencies.  Does not automatically
+         resolve cycles, and will raise :class:`.CircularDependencyError`
+         if cycles exist.
+
+         .. deprecated:: 1.0.0 - see
+            :meth:`.Inspector.get_sorted_table_and_fkc_names` for a version
+            of this which resolves foreign key cycles between tables
+            automatically.
 
          .. versionchanged:: 0.8 the "foreign_key" sorting sorts tables
             in order of dependee to dependent; that is, in creation
@@ -183,6 +190,8 @@ class Inspector(object):
 
         .. seealso::
 
+            :meth:`.Inspector.get_sorted_table_and_fkc_names`
+
             :attr:`.MetaData.sorted_tables`
 
         """
@@ -201,6 +210,64 @@ class Inspector(object):
             tnames = list(topological.sort(tuples, tnames))
         return tnames
 
+    def get_sorted_table_and_fkc_names(self, schema=None):
+        """Return dependency-sorted table and foreign key constraint names in
+        referred to within a particular schema.
+
+        This will yield 2-tuples of
+        ``(tablename, [(tname, fkname), (tname, fkname), ...])``
+        consisting of table names in CREATE order grouped with the foreign key
+        constraint names that are not detected as belonging to a cycle.
+        The final element
+        will be ``(None, [(tname, fkname), (tname, fkname), ..])``
+        which will consist of remaining
+        foreign key constraint names that would require a separate CREATE
+        step after-the-fact, based on dependencies between tables.
+
+        .. versionadded:: 1.0.-
+
+        .. seealso::
+
+            :meth:`.Inspector.get_table_names`
+
+            :func:`.sort_tables_and_constraints` - similar method which works
+             with an already-given :class:`.MetaData`.
+
+        """
+        if hasattr(self.dialect, 'get_table_names'):
+            tnames = self.dialect.get_table_names(
+                self.bind, schema, info_cache=self.info_cache)
+        else:
+            tnames = self.engine.table_names(schema)
+
+        tuples = set()
+        remaining_fkcs = set()
+
+        fknames_for_table = {}
+        for tname in tnames:
+            fkeys = self.get_foreign_keys(tname, schema)
+            fknames_for_table[tname] = set(
+                [fk['name'] for fk in fkeys]
+            )
+            for fkey in fkeys:
+                if tname != fkey['referred_table']:
+                    tuples.add((fkey['referred_table'], tname))
+        try:
+            candidate_sort = list(topological.sort(tuples, tnames))
+        except exc.CircularDependencyError as err:
+            for edge in err.edges:
+                tuples.remove(edge)
+                remaining_fkcs.update(
+                    (edge[1], fkc)
+                    for fkc in fknames_for_table[edge[1]]
+                )
+
+            candidate_sort = list(topological.sort(tuples, tnames))
+        return [
+            (tname, fknames_for_table[tname].difference(remaining_fkcs))
+            for tname in candidate_sort
+        ] + [(None, list(remaining_fkcs))]
+
     def get_temp_table_names(self):
         """return a list of temporary table names for the current bind.
 
diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py
index 95ebd05db..285ae579f 100644
--- a/lib/sqlalchemy/schema.py
+++ b/lib/sqlalchemy/schema.py
@@ -59,5 +59,7 @@ from .sql.ddl import (
     DDLBase,
     DDLElement,
     _CreateDropBase,
-    _DDLCompiles
+    _DDLCompiles,
+    sort_tables,
+    sort_tables_and_constraints
 )
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 9304bba9f..ca14c9371 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -2102,7 +2102,9 @@ class DDLCompiler(Compiled):
                         (table.description, column.name, ce.args[0])
                     ))
 
-        const = self.create_table_constraints(table)
+        const = self.create_table_constraints(
+            table, _include_foreign_key_constraints=
+            create.include_foreign_key_constraints)
         if const:
             text += ", \n\t" + const
 
@@ -2126,7 +2128,9 @@ class DDLCompiler(Compiled):
 
         return text
 
-    def create_table_constraints(self, table):
+    def create_table_constraints(
+        self, table,
+            _include_foreign_key_constraints=None):
 
         # On some DB order is significant: visit PK first, then the
         # other constraints (engine.ReflectionTest.testbasic failed on FB2)
@@ -2134,8 +2138,15 @@ class DDLCompiler(Compiled):
         if table.primary_key:
             constraints.append(table.primary_key)
 
+        all_fkcs = table.foreign_key_constraints
+        if _include_foreign_key_constraints is not None:
+            omit_fkcs = all_fkcs.difference(_include_foreign_key_constraints)
+        else:
+            omit_fkcs = set()
+
         constraints.extend([c for c in table._sorted_constraints
-                            if c is not table.primary_key])
+                            if c is not table.primary_key and
+                            c not in omit_fkcs])
 
         return ", \n\t".join(
             p for p in
@@ -2230,9 +2241,19 @@ class DDLCompiler(Compiled):
             self.preparer.format_sequence(drop.element)
 
     def visit_drop_constraint(self, drop):
+        constraint = drop.element
+        if constraint.name is not None:
+            formatted_name = self.preparer.format_constraint(constraint)
+        else:
+            formatted_name = None
+
+        if formatted_name is None:
+            raise exc.CompileError(
+                "Can't emit DROP CONSTRAINT for constraint %r; "
+                "it has no name" % drop.element)
         return "ALTER TABLE %s DROP CONSTRAINT %s%s" % (
             self.preparer.format_table(drop.element.table),
-            self.preparer.format_constraint(drop.element),
+            formatted_name,
             drop.cascade and " CASCADE" or ""
         )
 
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index 534322c8d..331a283f0 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -12,7 +12,6 @@ to invoke them for a create/drop call.
 
 from .. import util
 from .elements import ClauseElement
-from .visitors import traverse
 from .base import Executable, _generative, SchemaVisitor, _bind_or_error
 from ..util import topological
 from .. import event
@@ -464,19 +463,28 @@ class CreateTable(_CreateDropBase):
 
     __visit_name__ = "create_table"
 
-    def __init__(self, element, on=None, bind=None):
+    def __init__(
+            self, element, on=None, bind=None,
+            include_foreign_key_constraints=None):
         """Create a :class:`.CreateTable` construct.
 
         :param element: a :class:`.Table` that's the subject
          of the CREATE
         :param on: See the description for 'on' in :class:`.DDL`.
         :param bind: See the description for 'bind' in :class:`.DDL`.
+        :param include_foreign_key_constraints: optional sequence of
+         :class:`.ForeignKeyConstraint` objects that will be included
+         inline within the CREATE construct; if omitted, all foreign key
+         constraints that do not specify use_alter=True are included.
+
+         .. versionadded:: 1.0.0
 
         """
         super(CreateTable, self).__init__(element, on=on, bind=bind)
         self.columns = [CreateColumn(column)
                         for column in element.columns
                         ]
+        self.include_foreign_key_constraints = include_foreign_key_constraints
 
 
 class _DropView(_CreateDropBase):
@@ -696,8 +704,10 @@ class SchemaGenerator(DDLBase):
             tables = self.tables
         else:
             tables = list(metadata.tables.values())
-        collection = [t for t in sort_tables(tables)
-                      if self._can_create_table(t)]
+
+        collection = sort_tables_and_constraints(
+            [t for t in tables if self._can_create_table(t)])
+
         seq_coll = [s for s in metadata._sequences.values()
                     if s.column is None and self._can_create_sequence(s)]
 
@@ -709,15 +719,23 @@ class SchemaGenerator(DDLBase):
         for seq in seq_coll:
             self.traverse_single(seq, create_ok=True)
 
-        for table in collection:
-            self.traverse_single(table, create_ok=True)
+        for table, fkcs in collection:
+            if table is not None:
+                self.traverse_single(
+                    table, create_ok=True,
+                    include_foreign_key_constraints=fkcs)
+            else:
+                for fkc in fkcs:
+                    self.traverse_single(fkc)
 
         metadata.dispatch.after_create(metadata, self.connection,
                                        tables=collection,
                                        checkfirst=self.checkfirst,
                                        _ddl_runner=self)
 
-    def visit_table(self, table, create_ok=False):
+    def visit_table(
+            self, table, create_ok=False,
+            include_foreign_key_constraints=None):
         if not create_ok and not self._can_create_table(table):
             return
 
@@ -729,7 +747,15 @@ class SchemaGenerator(DDLBase):
             if column.default is not None:
                 self.traverse_single(column.default)
 
-        self.connection.execute(CreateTable(table))
+        if not self.dialect.supports_alter:
+            # e.g., don't omit any foreign key constraints
+            include_foreign_key_constraints = None
+
+        self.connection.execute(
+            CreateTable(
+                table,
+                include_foreign_key_constraints=include_foreign_key_constraints
+            ))
 
         if hasattr(table, 'indexes'):
             for index in table.indexes:
@@ -739,6 +765,11 @@ class SchemaGenerator(DDLBase):
                                     checkfirst=self.checkfirst,
                                     _ddl_runner=self)
 
+    def visit_foreign_key_constraint(self, constraint):
+        if not self.dialect.supports_alter:
+            return
+        self.connection.execute(AddConstraint(constraint))
+
     def visit_sequence(self, sequence, create_ok=False):
         if not create_ok and not self._can_create_sequence(sequence):
             return
@@ -765,11 +796,33 @@ class SchemaDropper(DDLBase):
         else:
             tables = list(metadata.tables.values())
 
-        collection = [
-            t
-            for t in reversed(sort_tables(tables))
-            if self._can_drop_table(t)
-        ]
+        try:
+            collection = reversed(
+                sort_tables_and_constraints(
+                    [t for t in tables if self._can_drop_table(t)],
+                    filter_fn=
+                    lambda constraint: True if not self.dialect.supports_alter
+                    else False if constraint.name is None
+                    else None
+                )
+            )
+        except exc.CircularDependencyError as err2:
+            util.raise_from_cause(
+                exc.CircularDependencyError(
+                    err2.message,
+                    err2.cycles, err2.edges,
+                    msg="Can't sort tables for DROP; an "
+                    "unresolvable foreign key "
+                    "dependency exists between tables: %s.  Please ensure "
+                    "that the ForeignKey and ForeignKeyConstraint objects "
+                    "involved in the cycle have "
+                    "names so that they can be dropped using DROP CONSTRAINT."
+                    % (
+                        ", ".join(sorted([t.fullname for t in err2.cycles]))
+                    )
+
+                )
+            )
 
         seq_coll = [
             s
@@ -781,8 +834,13 @@ class SchemaDropper(DDLBase):
             metadata, self.connection, tables=collection,
             checkfirst=self.checkfirst, _ddl_runner=self)
 
-        for table in collection:
-            self.traverse_single(table, drop_ok=True)
+        for table, fkcs in collection:
+            if table is not None:
+                self.traverse_single(
+                    table, drop_ok=True)
+            else:
+                for fkc in fkcs:
+                    self.traverse_single(fkc)
 
         for seq in seq_coll:
             self.traverse_single(seq, drop_ok=True)
@@ -830,6 +888,11 @@ class SchemaDropper(DDLBase):
                                   checkfirst=self.checkfirst,
                                   _ddl_runner=self)
 
+    def visit_foreign_key_constraint(self, constraint):
+        if not self.dialect.supports_alter:
+            return
+        self.connection.execute(DropConstraint(constraint))
+
     def visit_sequence(self, sequence, drop_ok=False):
         if not drop_ok and not self._can_drop_sequence(sequence):
             return
@@ -837,32 +900,159 @@ class SchemaDropper(DDLBase):
 
 
 def sort_tables(tables, skip_fn=None, extra_dependencies=None):
-    """sort a collection of Table objects in order of
-                their foreign-key dependency."""
+    """sort a collection of :class:`.Table` objects based on dependency.
 
-    tables = list(tables)
-    tuples = []
-    if extra_dependencies is not None:
-        tuples.extend(extra_dependencies)
+    This is a dependency-ordered sort which will emit :class:`.Table`
+    objects such that they will follow their dependent :class:`.Table` objects.
+    Tables are dependent on another based on the presence of
+    :class:`.ForeignKeyConstraint` objects as well as explicit dependencies
+    added by :meth:`.Table.add_is_dependent_on`.
+
+    .. warning::
+
+        The :func:`.sort_tables` function cannot by itself accommodate
+        automatic resolution of dependency cycles between tables, which
+        are usually caused by mutually dependent foreign key constraints.
+        To resolve these cycles, either the
+        :paramref:`.ForeignKeyConstraint.use_alter` parameter may be appled
+        to those constraints, or use the
+        :func:`.sql.sort_tables_and_constraints` function which will break
+        out foreign key constraints involved in cycles separately.
+
+    :param tables: a sequence of :class:`.Table` objects.
+
+    :param skip_fn: optional callable which will be passed a
+     :class:`.ForeignKey` object; if it returns True, this
+     constraint will not be considered as a dependency.  Note this is
+     **different** from the same parameter in
+     :func:`.sort_tables_and_constraints`, which is
+     instead passed the owning :class:`.ForeignKeyConstraint` object.
+
+    :param extra_dependencies: a sequence of 2-tuples of tables which will
+     also be considered as dependent on each other.
+
+    .. seealso::
+
+        :func:`.sort_tables_and_constraints`
+
+        :meth:`.MetaData.sorted_tables` - uses this function to sort
+
+
+    """
+
+    if skip_fn is not None:
+        def _skip_fn(fkc):
+            for fk in fkc.elements:
+                if skip_fn(fk):
+                    return True
+            else:
+                return None
+    else:
+        _skip_fn = None
+
+    return [
+        t for (t, fkcs) in
+        sort_tables_and_constraints(
+            tables, filter_fn=_skip_fn, extra_dependencies=extra_dependencies)
+        if t is not None
+    ]
+
+
+def sort_tables_and_constraints(
+        tables, filter_fn=None, extra_dependencies=None):
+    """sort a collection of :class:`.Table`  / :class:`.ForeignKeyConstraint`
+    objects.
+
+    This is a dependency-ordered sort which will emit tuples of
+    ``(Table, [ForeignKeyConstraint, ...])`` such that each
+    :class:`.Table` follows its dependent :class:`.Table` objects.
+    Remaining :class:`.ForeignKeyConstraint` objects that are separate due to
+    dependency rules not satisifed by the sort are emitted afterwards
+    as ``(None, [ForeignKeyConstraint ...])``.
+
+    Tables are dependent on another based on the presence of
+    :class:`.ForeignKeyConstraint` objects, explicit dependencies
+    added by :meth:`.Table.add_is_dependent_on`, as well as dependencies
+    stated here using the :paramref:`~.sort_tables_and_constraints.skip_fn`
+    and/or :paramref:`~.sort_tables_and_constraints.extra_dependencies`
+    parameters.
+
+    :param tables: a sequence of :class:`.Table` objects.
+
+    :param filter_fn: optional callable which will be passed a
+     :class:`.ForeignKeyConstraint` object, and returns a value based on
+     whether this constraint should definitely be included or excluded as
+     an inline constraint, or neither.   If it returns False, the constraint
+     will definitely be included as a dependency that cannot be subject
+     to ALTER; if True, it will **only** be included as an ALTER result at
+     the end.   Returning None means the constraint is included in the
+     table-based result unless it is detected as part of a dependency cycle.
+
+    :param extra_dependencies: a sequence of 2-tuples of tables which will
+     also be considered as dependent on each other.
+
+    .. versionadded:: 1.0.0
+
+    .. seealso::
+
+        :func:`.sort_tables`
 
-    def visit_foreign_key(fkey):
-        if fkey.use_alter:
-            return
-        elif skip_fn and skip_fn(fkey):
-            return
-        parent_table = fkey.column.table
-        if parent_table in tables:
-            child_table = fkey.parent.table
-            if parent_table is not child_table:
-                tuples.append((parent_table, child_table))
 
+    """
+
+    fixed_dependencies = set()
+    mutable_dependencies = set()
+
+    if extra_dependencies is not None:
+        fixed_dependencies.update(extra_dependencies)
+
+    remaining_fkcs = set()
     for table in tables:
-        traverse(table,
-                 {'schema_visitor': True},
-                 {'foreign_key': visit_foreign_key})
+        for fkc in table.foreign_key_constraints:
+            if fkc.use_alter is True:
+                remaining_fkcs.add(fkc)
+                continue
+
+            if filter_fn:
+                filtered = filter_fn(fkc)
+
+                if filtered is True:
+                    remaining_fkcs.add(fkc)
+                    continue
 
-        tuples.extend(
-            [parent, table] for parent in table._extra_dependencies
+            dependent_on = fkc.referred_table
+            if dependent_on is not table:
+                mutable_dependencies.add((dependent_on, table))
+
+        fixed_dependencies.update(
+            (parent, table) for parent in table._extra_dependencies
+        )
+
+    try:
+        candidate_sort = list(
+            topological.sort(
+                fixed_dependencies.union(mutable_dependencies), tables
+            )
+        )
+    except exc.CircularDependencyError as err:
+        for edge in err.edges:
+            if edge in mutable_dependencies:
+                table = edge[1]
+                can_remove = [
+                    fkc for fkc in table.foreign_key_constraints
+                    if filter_fn is None or filter_fn(fkc) is not False]
+                remaining_fkcs.update(can_remove)
+                for fkc in can_remove:
+                    dependent_on = fkc.referred_table
+                    if dependent_on is not table:
+                        mutable_dependencies.discard((dependent_on, table))
+        candidate_sort = list(
+            topological.sort(
+                fixed_dependencies.union(mutable_dependencies), tables
+            )
         )
 
-    return list(topological.sort(tuples, tables))
+    return [
+        (table, table.foreign_key_constraints.difference(remaining_fkcs))
+        for table in candidate_sort
+    ] + [(None, list(remaining_fkcs))]
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 71a0c2780..65a1da877 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1476,7 +1476,14 @@ class ForeignKey(DialectKWArgs, SchemaItem):
         :param use_alter: passed to the underlying
             :class:`.ForeignKeyConstraint` to indicate the constraint should
             be generated/dropped externally from the CREATE TABLE/ DROP TABLE
-            statement. See that classes' constructor for details.
+            statement.  See :paramref:`.ForeignKeyConstraint.use_alter`
+            for further description.
+
+            .. seealso::
+
+                :paramref:`.ForeignKeyConstraint.use_alter`
+
+                :ref:`use_alter`
 
         :param match: Optional string. If set, emit MATCH <value> when issuing
             DDL for this constraint. Typical values include SIMPLE, PARTIAL
@@ -2566,11 +2573,23 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
           part of the CREATE TABLE definition. Instead, generate it via an
           ALTER TABLE statement issued after the full collection of tables
           have been created, and drop it via an ALTER TABLE statement before
-          the full collection of tables are dropped. This is shorthand for the
-          usage of :class:`.AddConstraint` and :class:`.DropConstraint`
-          applied as "after-create" and "before-drop" events on the MetaData
-          object.  This is normally used to generate/drop constraints on
-          objects that are mutually dependent on each other.
+          the full collection of tables are dropped.
+
+          The use of :paramref:`.ForeignKeyConstraint.use_alter` is
+          particularly geared towards the case where two or more tables
+          are established within a mutually-dependent foreign key constraint
+          relationship; however, the :meth:`.MetaData.create_all` and
+          :meth:`.MetaData.drop_all` methods will perform this resolution
+          automatically, so the flag is normally not needed.
+
+          .. versionchanged:: 1.0.0  Automatic resolution of foreign key
+             cycles has been added, removing the need to use the
+             :paramref:`.ForeignKeyConstraint.use_alter` in typical use
+             cases.
+
+          .. seealso::
+
+                :ref:`use_alter`
 
         :param match: Optional string. If set, emit MATCH <value> when issuing
           DDL for this constraint. Typical values include SIMPLE, PARTIAL
@@ -2596,8 +2615,6 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
         self.onupdate = onupdate
         self.ondelete = ondelete
         self.link_to_name = link_to_name
-        if self.name is None and use_alter:
-            raise exc.ArgumentError("Alterable Constraint requires a name")
         self.use_alter = use_alter
         self.match = match
 
@@ -2648,7 +2665,7 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
     @property
     def referred_table(self):
         """The :class:`.Table` object to which this
-        :class:`.ForeignKeyConstraint references.
+        :class:`.ForeignKeyConstraint` references.
 
         This is a dynamically calculated attribute which may not be available
         if the constraint and/or parent table is not yet associated with
@@ -2716,15 +2733,6 @@ class ForeignKeyConstraint(ColumnCollectionConstraint):
 
         self._validate_dest_table(table)
 
-        if self.use_alter:
-            def supports_alter(ddl, event, schema_item, bind, **kw):
-                return table in set(kw['tables']) and \
-                    bind.dialect.supports_alter
-
-            event.listen(table.metadata, "after_create",
-                         ddl.AddConstraint(self, on=supports_alter))
-            event.listen(table.metadata, "before_drop",
-                         ddl.DropConstraint(self, on=supports_alter))
 
     def copy(self, schema=None, target_table=None, **kw):
         fkc = ForeignKeyConstraint(
@@ -3368,12 +3376,30 @@ class MetaData(SchemaItem):
         order in which they can be created.   To get the order in which
         the tables would be dropped, use the ``reversed()`` Python built-in.
 
+        .. warning::
+
+            The :attr:`.sorted_tables` accessor cannot by itself accommodate
+            automatic resolution of dependency cycles between tables, which
+            are usually caused by mutually dependent foreign key constraints.
+            To resolve these cycles, either the
+            :paramref:`.ForeignKeyConstraint.use_alter` parameter may be appled
+            to those constraints, or use the
+            :func:`.schema.sort_tables_and_constraints` function which will break
+            out foreign key constraints involved in cycles separately.
+
         .. seealso::
 
+            :func:`.schema.sort_tables`
+
+            :func:`.schema.sort_tables_and_constraints`
+
             :attr:`.MetaData.tables`
 
             :meth:`.Inspector.get_table_names`
 
+            :meth:`.Inspector.get_sorted_table_and_fkc_names`
+
+
         """
         return ddl.sort_tables(self.tables.values())
 
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 1f37b4b45..2375a13a9 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -23,7 +23,8 @@ from .assertions import emits_warning, emits_warning_on, uses_deprecated, \
     assert_raises_message, AssertsCompiledSQL, ComparesTables, \
     AssertsExecutionResults, expect_deprecated, expect_warnings
 
-from .util import run_as_contextmanager, rowset, fail, provide_metadata, adict
+from .util import run_as_contextmanager, rowset, fail, \
+    provide_metadata, adict, force_drop_names
 
 crashes = skip
 
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index 614a12133..646e4dea2 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -325,19 +325,11 @@ def _prep_testing_database(options, file_config):
                                          schema="test_schema")
                         ))
 
-            for tname in reversed(inspector.get_table_names(
-                    order_by="foreign_key")):
-                e.execute(schema.DropTable(
-                    schema.Table(tname, schema.MetaData())
-                ))
+            util.drop_all_tables(e, inspector)
 
             if config.requirements.schemas.enabled_for_config(cfg):
-                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")
-                    ))
+                util.drop_all_tables(e, inspector, schema=cfg.test_schema)
+                util.drop_all_tables(e, inspector, schema=cfg.test_schema_2)
 
             if against(cfg, "postgresql"):
                 from sqlalchemy.dialects import postgresql
diff --git a/lib/sqlalchemy/testing/util.py b/lib/sqlalchemy/testing/util.py
index 7b3f721a6..eea39b1f7 100644
--- a/lib/sqlalchemy/testing/util.py
+++ b/lib/sqlalchemy/testing/util.py
@@ -194,6 +194,25 @@ def provide_metadata(fn, *args, **kw):
         self.metadata = prev_meta
 
 
+def force_drop_names(*names):
+    """Force the given table names to be dropped after test complete,
+    isolating for foreign key cycles
+
+    """
+    from . import config
+    from sqlalchemy import inspect
+
+    @decorator
+    def go(fn, *args, **kw):
+
+        try:
+            return fn(*args, **kw)
+        finally:
+            drop_all_tables(
+                config.db, inspect(config.db), include_names=names)
+    return go
+
+
 class adict(dict):
     """Dict keys available as attributes.  Shadows."""
 
@@ -207,3 +226,39 @@ class adict(dict):
         return tuple([self[key] for key in keys])
 
     get_all = __call__
+
+
+def drop_all_tables(engine, inspector, schema=None, include_names=None):
+    from sqlalchemy import Column, Table, Integer, MetaData, \
+        ForeignKeyConstraint
+    from sqlalchemy.schema import DropTable, DropConstraint
+
+    if include_names is not None:
+        include_names = set(include_names)
+
+    with engine.connect() as conn:
+        for tname, fkcs in reversed(
+                inspector.get_sorted_table_and_fkc_names(schema=schema)):
+            if tname:
+                if include_names is not None and tname not in include_names:
+                    continue
+                conn.execute(DropTable(
+                    Table(tname, MetaData())
+                ))
+            elif fkcs:
+                if not engine.dialect.supports_alter:
+                    continue
+                for tname, fkc in fkcs:
+                    if include_names is not None and \
+                            tname not in include_names:
+                        continue
+                    tb = Table(
+                        tname, MetaData(),
+                        Column('x', Integer),
+                        Column('y', Integer),
+                        schema=schema
+                    )
+                    conn.execute(DropConstraint(
+                        ForeignKeyConstraint(
+                            [tb.c.x], [tb.c.y], name=fkc)
+                    ))