- The Postgresql :class:`.postgresql.ENUM` type will emit a

DROP TYPE instruction when a plain ``table.drop()`` is called,
assuming the object is not associated directly with a
:class:`.MetaData` object.   In order to accomodate the use case of
an enumerated type shared between multiple tables, the type should
be associated directly with the :class:`.MetaData` object; in this
case the type will only be created at the metadata level, or if
created directly.  The rules for create/drop of
Postgresql enumerated types have been highly reworked in general.
fixes #3319

1 files changed, 26 insertions, 16 deletions

diff --git a/lib/sqlalchemy/sql/ddl.py b/lib/sqlalchemy/sql/ddl.py
index d6c5f1253..3834f25f4 100644
--- a/lib/sqlalchemy/sql/ddl.py
+++ b/lib/sqlalchemy/sql/ddl.py
@@ -723,7 +723,8 @@ class SchemaGenerator(DDLBase):
             if table is not None:
                 self.traverse_single(
                     table, create_ok=True,
-                    include_foreign_key_constraints=fkcs)
+                    include_foreign_key_constraints=fkcs,
+                    _is_metadata_operation=True)
             else:
                 for fkc in fkcs:
                     self.traverse_single(fkc)
@@ -735,13 +736,16 @@ class SchemaGenerator(DDLBase):
 
     def visit_table(
             self, table, create_ok=False,
-            include_foreign_key_constraints=None):
+            include_foreign_key_constraints=None,
+            _is_metadata_operation=False):
         if not create_ok and not self._can_create_table(table):
             return
 
-        table.dispatch.before_create(table, self.connection,
-                                     checkfirst=self.checkfirst,
-                                     _ddl_runner=self)
+        table.dispatch.before_create(
+            table, self.connection,
+            checkfirst=self.checkfirst,
+            _ddl_runner=self,
+            _is_metadata_operation=_is_metadata_operation)
 
         for column in table.columns:
             if column.default is not None:
@@ -761,9 +765,11 @@ class SchemaGenerator(DDLBase):
             for index in table.indexes:
                 self.traverse_single(index)
 
-        table.dispatch.after_create(table, self.connection,
-                                    checkfirst=self.checkfirst,
-                                    _ddl_runner=self)
+        table.dispatch.after_create(
+            table, self.connection,
+            checkfirst=self.checkfirst,
+            _ddl_runner=self,
+            _is_metadata_operation=_is_metadata_operation)
 
     def visit_foreign_key_constraint(self, constraint):
         if not self.dialect.supports_alter:
@@ -837,7 +843,7 @@ class SchemaDropper(DDLBase):
         for table, fkcs in collection:
             if table is not None:
                 self.traverse_single(
-                    table, drop_ok=True)
+                    table, drop_ok=True, _is_metadata_operation=True)
             else:
                 for fkc in fkcs:
                     self.traverse_single(fkc)
@@ -870,13 +876,15 @@ class SchemaDropper(DDLBase):
     def visit_index(self, index):
         self.connection.execute(DropIndex(index))
 
-    def visit_table(self, table, drop_ok=False):
+    def visit_table(self, table, drop_ok=False, _is_metadata_operation=False):
         if not drop_ok and not self._can_drop_table(table):
             return
 
-        table.dispatch.before_drop(table, self.connection,
-                                   checkfirst=self.checkfirst,
-                                   _ddl_runner=self)
+        table.dispatch.before_drop(
+            table, self.connection,
+            checkfirst=self.checkfirst,
+            _ddl_runner=self,
+            _is_metadata_operation=_is_metadata_operation)
 
         for column in table.columns:
             if column.default is not None:
@@ -884,9 +892,11 @@ class SchemaDropper(DDLBase):
 
         self.connection.execute(DropTable(table))
 
-        table.dispatch.after_drop(table, self.connection,
-                                  checkfirst=self.checkfirst,
-                                  _ddl_runner=self)
+        table.dispatch.after_drop(
+            table, self.connection,
+           checkfirst=self.checkfirst,
+           _ddl_runner=self,
+           _is_metadata_operation=_is_metadata_operation)
 
     def visit_foreign_key_constraint(self, constraint):
         if not self.dialect.supports_alter: