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: https://github.com/sqlalchemy/alembic/pull/1036
Pull-request-sha: ea44e7f3398aa5001e86e6a7e30f61f6a86d9674

Change-Id: I20a1702e17ed88054206d964152ce05b81d0f89e

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/docs/build/unreleased/1037.rst b/docs/build/unreleased/1037.rst
new file mode 100644 (file)
index 0000000..12ef882
--- /dev/null
+++ b/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.

diff --git a/tests/test_op.py b/tests/test_op.py
index 677b7bc05b9e11021599eb91da29b4ab8c3b4fb8..5682abb8e7968952f7601ac78002f8358089b128 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,58 @@ 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):