- 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

4 files changed, 141 insertions, 40 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index b770569d0..b8392bd4e 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1382,6 +1382,7 @@ class ENUM(sqltypes.Enum, _EnumeratedValues):
         kw.pop('quote', None)
         kw.pop('native_enum', None)
         kw.pop('inherit_schema', None)
+        kw.pop('_create_events', None)
         _StringType.__init__(self, length=length, **kw)
         sqltypes.Enum.__init__(self, *values)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 8e18ca7e4..7529c6ed3 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -495,6 +495,22 @@ dialect in conjunction with the :class:`.Table` construct:
     `Postgresql CREATE TABLE options
     <http://www.postgresql.org/docs/9.3/static/sql-createtable.html>`_
 
+ENUM Types
+----------
+
+Postgresql has an independently creatable TYPE structure which is used
+to implement an enumerated type.   This approach introduces significant
+complexity on the SQLAlchemy side in terms of when this type should be
+CREATED and DROPPED.   The type object is also an independently reflectable
+entity.   The following sections should be consulted:
+
+* :class:`.postgresql.ENUM` - DDL and typing support for ENUM.
+
+* :meth:`.PGInspector.get_enums` - retrieve a listing of current ENUM types
+
+* :meth:`.postgresql.ENUM.create` , :meth:`.postgresql.ENUM.drop` - individual
+  CREATE and DROP commands for ENUM.
+
 """
 from collections import defaultdict
 import re
@@ -1099,21 +1115,76 @@ class ENUM(sqltypes.Enum):
     """Postgresql ENUM type.
 
     This is a subclass of :class:`.types.Enum` which includes
-    support for PG's ``CREATE TYPE``.
-
-    :class:`~.postgresql.ENUM` is used automatically when
-    using the :class:`.types.Enum` type on PG assuming
-    the ``native_enum`` is left as ``True``.   However, the
-    :class:`~.postgresql.ENUM` class can also be instantiated
-    directly in order to access some additional Postgresql-specific
-    options, namely finer control over whether or not
-    ``CREATE TYPE`` should be emitted.
-
-    Note that both :class:`.types.Enum` as well as
-    :class:`~.postgresql.ENUM` feature create/drop
-    methods; the base :class:`.types.Enum` type ultimately
-    delegates to the :meth:`~.postgresql.ENUM.create` and
-    :meth:`~.postgresql.ENUM.drop` methods present here.
+    support for PG's ``CREATE TYPE`` and ``DROP TYPE``.
+
+    When the builtin type :class:`.types.Enum` is used and the
+    :paramref:`.Enum.native_enum` flag is left at its default of
+    True, the Postgresql backend will use a :class:`.postgresql.ENUM`
+    type as the implementation, so the special create/drop rules
+    will be used.
+
+    The create/drop behavior of ENUM is necessarily intricate, due to the
+    awkward relationship the ENUM type has in relationship to the
+    parent table, in that it may be "owned" by just a single table, or
+    may be shared among many tables.
+
+    When using :class:`.types.Enum` or :class:`.postgresql.ENUM`
+    in an "inline" fashion, the ``CREATE TYPE`` and ``DROP TYPE`` is emitted
+    corresponding to when the :meth:`.Table.create` and :meth:`.Table.drop`
+    methods are called::
+
+        table = Table('sometable', metadata,
+            Column('some_enum', ENUM('a', 'b', 'c', name='myenum'))
+        )
+
+        table.create(engine)  # will emit CREATE ENUM and CREATE TABLE
+        table.drop(engine)  # will emit DROP TABLE and DROP ENUM
+
+    To use a common enumerated type between multiple tables, the best
+    practice is to declare the :class:`.types.Enum` or
+    :class:`.postgresql.ENUM` independently, and associate it with the
+    :class:`.MetaData` object itself::
+
+        my_enum = ENUM('a', 'b', 'c', name='myenum', metadata=metadata)
+
+        t1 = Table('sometable_one', metadata,
+            Column('some_enum', myenum)
+        )
+
+        t2 = Table('sometable_two', metadata,
+            Column('some_enum', myenum)
+        )
+
+    When this pattern is used, care must still be taken at the level
+    of individual table creates.  Emitting CREATE TABLE without also
+    specifying ``checkfirst=True`` will still cause issues::
+
+        t1.create(engine) # will fail: no such type 'myenum'
+
+    If we specify ``checkfirst=True``, the individual table-level create
+    operation will check for the ``ENUM`` and create if not exists::
+
+        # will check if enum exists, and emit CREATE TYPE if not
+        t1.create(engine, checkfirst=True)
+
+    When using a metadata-level ENUM type, the type will always be created
+    and dropped if either the metadata-wide create/drop is called::
+
+        metadata.create_all(engine)  # will emit CREATE TYPE
+        metadata.drop_all(engine)  # will emit DROP TYPE
+
+    The type can also be created and dropped directly::
+
+        my_enum.create(engine)
+        my_enum.drop(engine)
+
+    .. versionchanged:: 1.0.0 The Postgresql :class:`.postgresql.ENUM` type
+       now behaves more strictly with regards to CREATE/DROP.  A metadata-level
+       ENUM type will only be created and dropped at the metadata level,
+       not the table level, with the exception of
+       ``table.create(checkfirst=True)``.
+       The ``table.drop()`` call will now emit a DROP TYPE for a table-level
+       enumerated type.
 
     """
 
@@ -1219,9 +1290,18 @@ class ENUM(sqltypes.Enum):
             return False
 
     def _on_table_create(self, target, bind, checkfirst, **kw):
-        if not self._check_for_name_in_memos(checkfirst, kw):
+        if checkfirst or (
+                not self.metadata and
+                not kw.get('_is_metadata_operation', False)) and \
+                not self._check_for_name_in_memos(checkfirst, kw):
             self.create(bind=bind, checkfirst=checkfirst)
 
+    def _on_table_drop(self, target, bind, checkfirst, **kw):
+        if not self.metadata and \
+            not kw.get('_is_metadata_operation', False) and \
+                not self._check_for_name_in_memos(checkfirst, kw):
+            self.drop(bind=bind, checkfirst=checkfirst)
+
     def _on_metadata_create(self, target, bind, checkfirst, **kw):
         if not self._check_for_name_in_memos(checkfirst, kw):
             self.create(bind=bind, checkfirst=checkfirst)
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