Trigger events before and after drop table statements

The ``op.drop_table()`` operation directive will now trigger the
``before_drop()`` and ``after_drop()`` DDL event hooks at the table level,
which is similar to how the ``before_create()`` and ``after_create()``
hooks are triggered by the ``op.create_table()`` directive. Note that as
``op.drop_table()`` accepts only a table name and optional schema name, the
``Table`` object received by the event will not have any information within
it other than the table name and schema name.

Fixes: #1037
Closes: #1036
Pull-request: #1036
Pull-request-sha: ea44e7f

Change-Id: I20a1702e17ed88054206d964152ce05b81d0f89e

alembic/ddl/impl.py

@@ -371,7 +371,13 @@ def create_table(self, table: "Table") -> None:
                 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))

docs/build/unreleased/1037.rst

@@ -0,0 +1,11 @@
+.. change::
+    :tags: usecase, operations
+    :tickets: 1037
+
+    The ``op.drop_table()`` operation directive will now trigger the
+    ``before_drop()`` and ``after_drop()`` DDL event hooks at the table level,
+    which is similar to how the ``before_create()`` and ``after_create()``
+    hooks are triggered by the ``op.create_table()`` directive. Note that as
+    ``op.drop_table()`` accepts only a table name and optional schema name, the
+    ``Table`` object received by the event will not have any information within
+    it other than the table name and schema name.

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,58 @@ def test_drop_table_comment_op(self):
 
         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):