- 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

3 files changed, 60 insertions, 12 deletions

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)
+                    ))