Trigger events before and after drop table statements

diff --git a/alembic/ddl/impl.py b/alembic/ddl/impl.py
index 8c9e0b91fb138f21ef405e6f34cbea52910b5d4e..070c124bdedf435efbc9b33b520ff90e9bf39de8 100644 (file)
--- a/alembic/ddl/impl.py
+++ b/alembic/ddl/impl.py
@@ -371,7 +371,13 @@ class DefaultImpl(metaclass=ImplMeta):
                 self.create_column_comment(column)
 
     def drop_table(self, table: "Table") -> None:
+        table.dispatch.before_drop(
+            table, self.connection, checkfirst=False, _ddl_runner=self
+        )
         self._exec(schema.DropTable(table))
+        table.dispatch.after_drop(
+            table, self.connection, checkfirst=False, _ddl_runner=self
+        )
 
     def create_index(self, index: "Index") -> None:
         self._exec(schema.CreateIndex(index))

diff --git a/tests/test_op.py b/tests/test_op.py
index 677b7bc05b9e11021599eb91da29b4ab8c3b4fb8..7ed7e45acf4a447c498e72860faf0394ac569e5c 100644 (file)
--- a/tests/test_op.py
+++ b/tests/test_op.py
@@ -3,6 +3,7 @@
 from sqlalchemy import Boolean
 from sqlalchemy import CheckConstraint
 from sqlalchemy import Column
+from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy import ForeignKey
 from sqlalchemy import Index
@@ -1087,6 +1088,54 @@ class OpTest(TestBase):
 
         context.assert_("COMMENT ON TABLE some_table IS NULL")
 
+    def test_create_table_event(self):
+        context = op_fixture()
+
+        events_triggered = []
+
+        TestTable = Table("tb_test", MetaData(), Column("c1", Integer, nullable=False))
+
+        @event.listens_for(Table, "before_create")
+        def record_before_event(table, conn, **kwargs):
+            events_triggered.append(("before_create", table.name))
+
+        @event.listens_for(Table, "after_create")
+        def record_after_event(table, conn, **kwargs):
+            events_triggered.append(("after_create", table.name))
+
+        op.create_table(TestTable)
+        op.drop_table(TestTable)
+        context.assert_("CREATE TABLE tb_test ()", "DROP TABLE tb_test")
+
+        assert events_triggered == [
+            ("before_create", "tb_test"),
+            ("after_create", "tb_test"),
+        ]
+
+    def test_drop_table_event(self):
+        context = op_fixture()
+
+        events_triggered = []
+
+        TestTable = Table("tb_test", MetaData(), Column("c1", Integer, nullable=False))
+
+        @event.listens_for(Table, "before_drop")
+        def record_before_event(table, conn, **kwargs):
+            events_triggered.append(("before_drop", table.name))
+
+        @event.listens_for(Table, "after_drop")
+        def record_after_event(table, conn, **kwargs):
+            events_triggered.append(("after_drop", table.name))
+
+        op.create_table(TestTable)
+        op.drop_table(TestTable)
+        context.assert_("CREATE TABLE tb_test ()", "DROP TABLE tb_test")
+
+        assert events_triggered == [
+            ("before_drop", "tb_test"),
+            ("after_drop", "tb_test"),
+        ]
+
 
 class SQLModeOpTest(TestBase):
     def test_auto_literals(self):