- Fixed regression due to :ticket:`3282` where the ``tables`` collection

passed as a keyword argument to the :meth:`.DDLEvents.before_create`,
:meth:`.DDLEvents.after_create`, :meth:`.DDLEvents.before_drop`, and
:meth:`.DDLEvents.after_drop` events would no longer be a list
of tables, but instead a list of tuples which contained a second
entry with foreign keys to be added or dropped.  As the ``tables``
collection, while documented as not necessarily stable, has come
to be relied upon, this change is considered a regression.
Additionally, in some cases for "drop", this collection would
be an iterator that would cause the operation to fail if
prematurely iterated.   The collection is now a list of table
objects in all cases and test coverage for the format of this
collection is now added.
fixes #3391

3 files changed, 85 insertions, 8 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 891981fe2..b313c7cac 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -19,6 +19,25 @@
     :version: 1.0.3
 
     .. change::
+        :tags: bug, sql
+        :tickets: 3391
+
+        Fixed regression due to :ticket:`3282` where the ``tables`` collection
+        passed as a keyword argument to the :meth:`.DDLEvents.before_create`,
+        :meth:`.DDLEvents.after_create`, :meth:`.DDLEvents.before_drop`, and
+        :meth:`.DDLEvents.after_drop` events would no longer be a list
+        of tables, but instead a list of tuples which contained a second
+        entry with foreign keys to be added or dropped.  As the ``tables``
+        collection, while documented as not necessarily stable, has come
+        to be relied upon, this change is considered a regression.
+        Additionally, in some cases for "drop", this collection would
+        be an iterator that would cause the operation to fail if
+        prematurely iterated.   The collection is now a list of table
+        objects in all cases and test coverage for the format of this
+        collection is now added.
+
+
+    .. change::
         :tags: bug, orm
         :tickets: 3388
 
diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index a0841b13c..71018f132 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -711,8 +711,11 @@ class SchemaGenerator(DDLBase):
         seq_coll = [s for s in metadata._sequences.values()
                     if s.column is None and self._can_create_sequence(s)]
 
+        event_collection = [
+            t for (t, fks) in collection if t is not None
+        ]
         metadata.dispatch.before_create(metadata, self.connection,
-                                        tables=collection,
+                                        tables=event_collection,
                                         checkfirst=self.checkfirst,
                                         _ddl_runner=self)
 
@@ -730,7 +733,7 @@ class SchemaGenerator(DDLBase):
                     self.traverse_single(fkc)
 
         metadata.dispatch.after_create(metadata, self.connection,
-                                       tables=collection,
+                                       tables=event_collection,
                                        checkfirst=self.checkfirst,
                                        _ddl_runner=self)
 
@@ -804,7 +807,7 @@ class SchemaDropper(DDLBase):
 
         try:
             unsorted_tables = [t for t in tables if self._can_drop_table(t)]
-            collection = reversed(
+            collection = list(reversed(
                 sort_tables_and_constraints(
                     unsorted_tables,
                     filter_fn=lambda constraint: False
@@ -812,7 +815,7 @@ class SchemaDropper(DDLBase):
                     or constraint.name is None
                     else None
                 )
-            )
+            ))
         except exc.CircularDependencyError as err2:
             if not self.dialect.supports_alter:
                 util.warn(
@@ -854,8 +857,12 @@ class SchemaDropper(DDLBase):
             if s.column is None and self._can_drop_sequence(s)
         ]
 
+        event_collection = [
+            t for (t, fks) in collection if t is not None
+        ]
+
         metadata.dispatch.before_drop(
-            metadata, self.connection, tables=collection,
+            metadata, self.connection, tables=event_collection,
             checkfirst=self.checkfirst, _ddl_runner=self)
 
         for table, fkcs in collection:
@@ -870,7 +877,7 @@ class SchemaDropper(DDLBase):
             self.traverse_single(seq, drop_ok=True)
 
         metadata.dispatch.after_drop(
-            metadata, self.connection, tables=collection,
+            metadata, self.connection, tables=event_collection,
             checkfirst=self.checkfirst, _ddl_runner=self)
 
     def _can_drop_table(self, table):
diff --git a/test/engine/test_ddlevents.py b/test/engine/test_ddlevents.py
index 18300179c..cb13daa4c 100644
--- a/test/engine/test_ddlevents.py
+++ b/test/engine/test_ddlevents.py
@@ -19,29 +19,45 @@ class DDLEventTest(fixtures.TestBase):
             self.state = None
             self.schema_item = schema_item
             self.bind = bind
+            if isinstance(schema_item, MetaData):
+                self.tables = set(schema_item.tables.values())
+            else:
+                self.tables = None
 
         def before_create(self, schema_item, bind, **kw):
             assert self.state is None
             assert schema_item is self.schema_item
             assert bind is self.bind
+            if self.tables:
+                eq_(self.tables, set(kw['tables']))
+                assert isinstance(kw['tables'], list)
             self.state = 'before-create'
 
         def after_create(self, schema_item, bind, **kw):
             assert self.state in ('before-create', 'skipped')
             assert schema_item is self.schema_item
             assert bind is self.bind
+            if self.tables:
+                eq_(self.tables, set(kw['tables']))
+                assert isinstance(kw['tables'], list)
             self.state = 'after-create'
 
         def before_drop(self, schema_item, bind, **kw):
-            assert self.state is None
+            assert self.state in (None, 'skipped')
             assert schema_item is self.schema_item
             assert bind is self.bind
+            if self.tables:
+                eq_(self.tables, set(kw['tables']))
+                assert isinstance(kw['tables'], list)
             self.state = 'before-drop'
 
         def after_drop(self, schema_item, bind, **kw):
             assert self.state in ('before-drop', 'skipped')
             assert schema_item is self.schema_item
             assert bind is self.bind
+            if self.tables:
+                eq_(self.tables, set(kw['tables']))
+                assert isinstance(kw['tables'], list)
             self.state = 'after-drop'
 
     def setup(self):
@@ -130,7 +146,7 @@ class DDLEventTest(fixtures.TestBase):
         table.drop(bind)
         assert canary.state == 'after-drop'
 
-    def test_table_create_before(self):
+    def test_metadata_create_before(self):
         metadata, bind = self.metadata, self.bind
         canary = self.Canary(metadata, bind)
         event.listen(metadata, 'before_create', canary.before_create)
@@ -163,6 +179,41 @@ class DDLEventTest(fixtures.TestBase):
         metadata.drop_all(bind)
         assert canary.state == 'after-create'
 
+    def test_metadata_drop_before(self):
+        metadata, bind = self.metadata, self.bind
+        canary = self.Canary(metadata, bind)
+        event.listen(metadata, 'before_drop', canary.before_drop)
+
+        canary.state = 'skipped'
+        metadata.create_all(bind)
+        assert canary.state == 'skipped'
+        metadata.drop_all(bind)
+        assert canary.state == 'before-drop'
+
+    def test_metadata_drop_after(self):
+        metadata, bind = self.metadata, self.bind
+        canary = self.Canary(metadata, bind)
+        event.listen(metadata, 'after_drop', canary.after_drop)
+
+        canary.state = 'skipped'
+        metadata.create_all(bind)
+        assert canary.state == 'skipped'
+        metadata.drop_all(bind)
+        assert canary.state == 'after-drop'
+
+    def test_metadata_drop_both(self):
+        metadata, bind = self.metadata, self.bind
+        canary = self.Canary(metadata, bind)
+
+        event.listen(metadata, 'before_drop', canary.before_drop)
+        event.listen(metadata, 'after_drop', canary.after_drop)
+
+        canary.state = 'skipped'
+        metadata.create_all(bind)
+        assert canary.state == 'skipped'
+        metadata.drop_all(bind)
+        assert canary.state == 'after-drop'
+
     def test_metadata_table_isolation(self):
         metadata, table, bind = self.metadata, self.table, self.bind
         table_canary = self.Canary(table, bind)