- 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

1 files changed, 55 insertions, 0 deletions

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