- 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

2 files changed, 44 insertions, 24 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:
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 107dcee61..7e2e601e2 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -938,7 +938,7 @@ class SchemaType(SchemaEventTarget):
     """
 
     def __init__(self, name=None, schema=None, metadata=None,
-                 inherit_schema=False, quote=None):
+                 inherit_schema=False, quote=None, _create_events=True):
         if name is not None:
             self.name = quoted_name(name, quote)
         else:
@@ -946,8 +946,9 @@ class SchemaType(SchemaEventTarget):
         self.schema = schema
         self.metadata = metadata
         self.inherit_schema = inherit_schema
+        self._create_events = _create_events
 
-        if self.metadata:
+        if _create_events and self.metadata:
             event.listen(
                 self.metadata,
                 "before_create",
@@ -966,6 +967,9 @@ class SchemaType(SchemaEventTarget):
         if self.inherit_schema:
             self.schema = table.schema
 
+        if not self._create_events:
+            return
+
         event.listen(
             table,
             "before_create",
@@ -992,17 +996,18 @@ class SchemaType(SchemaEventTarget):
             )
 
     def copy(self, **kw):
-        return self.adapt(self.__class__)
+        return self.adapt(self.__class__, _create_events=True)
 
     def adapt(self, impltype, **kw):
         schema = kw.pop('schema', self.schema)
+        metadata = kw.pop('metadata', self.metadata)
+        _create_events = kw.pop('_create_events', False)
 
-        # don't associate with self.metadata as the hosting type
-        # is already associated with it, avoid creating event
-        # listeners
         return impltype(name=self.name,
                         schema=schema,
                         inherit_schema=self.inherit_schema,
+                        metadata=metadata,
+                        _create_events=_create_events,
                         **kw)
 
     @property
@@ -1170,7 +1175,8 @@ class Enum(String, SchemaType):
 
     def adapt(self, impltype, **kw):
         schema = kw.pop('schema', self.schema)
-        metadata = kw.pop('metadata', None)
+        metadata = kw.pop('metadata', self.metadata)
+        _create_events = kw.pop('_create_events', False)
         if issubclass(impltype, Enum):
             return impltype(name=self.name,
                             schema=schema,
@@ -1178,9 +1184,11 @@ class Enum(String, SchemaType):
                             convert_unicode=self.convert_unicode,
                             native_enum=self.native_enum,
                             inherit_schema=self.inherit_schema,
+                            _create_events=_create_events,
                             *self.enums,
                             **kw)
         else:
+            # TODO: why would we be here?
             return super(Enum, self).adapt(impltype, **kw)
 
 
@@ -1276,7 +1284,8 @@ class Boolean(TypeEngine, SchemaType):
 
     __visit_name__ = 'boolean'
 
-    def __init__(self, create_constraint=True, name=None):
+    def __init__(
+            self, create_constraint=True, name=None, _create_events=True):
         """Construct a Boolean.
 
         :param create_constraint: defaults to True.  If the boolean
@@ -1289,6 +1298,7 @@ class Boolean(TypeEngine, SchemaType):
         """
         self.create_constraint = create_constraint
         self.name = name
+        self._create_events = _create_events
 
     def _should_create_constraint(self, compiler):
         return not compiler.dialect.supports_native_boolean