- Added new ENUM type to the Postgresql dialect, which exists as a schema-level

construct and extends the generic Enum type.  Automatically
associates itself with tables and their parent metadata
to issue the appropriate CREATE TYPE/DROP TYPE
commands as needed, supports unicode labels, supports
reflection.  [ticket:1511]

- MySQL ENUM now subclasses the new generic Enum type, and also handles
unicode values implicitly, if the given labelnames are unicode
objects.

- Added a new Enum generic type, currently supported on
Postgresql and MySQL.  Enum is a schema-aware object
to support databases which require specific DDL in
order to use enum or equivalent; in the case of PG
it handles the details of `CREATE TYPE`, and on
other databases without native enum support can
support generation of CHECK constraints.
[ticket:1109] [ticket:1511]

- types documentation updates

- some cleanup on schema/expression docs

7 files changed, 383 insertions, 84 deletions

diff --git a/lib/sqlalchemy/__init__.py b/lib/sqlalchemy/__init__.py
index 545214fcd..4251732ee 100644
--- a/lib/sqlalchemy/__init__.py
+++ b/lib/sqlalchemy/__init__.py
@@ -60,6 +60,7 @@ from sqlalchemy.types import (
     DECIMAL,
     Date,
     DateTime,
+    Enum,
     FLOAT,
     Float,
     INT,
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 9b6671ece..d7ea358b5 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -876,7 +876,7 @@ class LONGBLOB(_BinaryType):
 
     __visit_name__ = 'LONGBLOB'
 
-class ENUM(_StringType):
+class ENUM(sqltypes.Enum, _StringType):
     """MySQL ENUM type."""
 
     __visit_name__ = 'ENUM'
@@ -955,20 +955,27 @@ class ENUM(_StringType):
                 'Manually quoting ENUM value literals is deprecated.  Supply '
                 'unquoted values and use the quoting= option in cases of '
                 'ambiguity.')
-            strip_enums = []
-            for a in enums:
-                if a[0:1] == '"' or a[0:1] == "'":
-                    # strip enclosing quotes and unquote interior
-                    a = a[1:-1].replace(a[0] * 2, a[0])
-                strip_enums.append(a)
-            self.enums = strip_enums
-        else:
-            self.enums = list(enums)
+            enums = self._strip_enums(enums)
 
         self.strict = kw.pop('strict', False)
-        length = max([len(v) for v in self.enums] + [0])
-        super(ENUM, self).__init__(length=length, **kw)
-
+        length = max([len(v) for v in enums] + [0])
+        kw.pop('metadata', None)
+        kw.pop('schema', None)
+        kw.pop('name', None)
+        kw.pop('quote', None)
+        _StringType.__init__(self, length=length, **kw)
+        sqltypes.Enum.__init__(self, *enums)
+    
+    @classmethod
+    def _strip_enums(cls, enums):
+        strip_enums = []
+        for a in enums:
+            if a[0:1] == '"' or a[0:1] == "'":
+                # strip enclosing quotes and unquote interior
+                a = a[1:-1].replace(a[0] * 2, a[0])
+            strip_enums.append(a)
+        return strip_enums
+        
     def bind_processor(self, dialect):
         super_convert = super(ENUM, self).bind_processor(dialect)
         def process(value):
@@ -1133,6 +1140,7 @@ colspecs = {
     sqltypes.Binary: _BinaryType,
     sqltypes.Boolean: _MSBoolean,
     sqltypes.Time: _MSTime,
+    sqltypes.Enum: ENUM,
 }
 
 # Everything 3.23 through 5.1 excepting OpenGIS types.
@@ -1566,7 +1574,10 @@ class MySQLTypeCompiler(compiler.GenericTypeCompiler):
     
     def visit_binary(self, type_):
         return self.visit_BLOB(type_)
-        
+    
+    def visit_enum(self, type_):
+        return self.visit_ENUM(type_)
+    
     def visit_BINARY(self, type_):
         if type_.length:
             return "BINARY(%d)" % type_.length
@@ -2249,7 +2260,7 @@ class MySQLTableDefinitionParser(object):
                 type_kw[kw] = spec[kw]
 
         if type_ == 'enum':
-            type_kw['quoting'] = 'quoted'
+            type_args = ENUM._strip_enums(type_args)
 
         type_instance = col_type(*type_args, **type_kw)
 
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 3c905e7d5..1f4858cdd 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -177,8 +177,29 @@ class ARRAY(sqltypes.MutableType, sqltypes.Concatenable, sqltypes.TypeEngine):
         return process
 PGArray = ARRAY
 
+class ENUM(sqltypes.Enum):
+
+    def create(self, bind=None, checkfirst=True):
+        if not checkfirst or not bind.dialect.has_type(bind, self.name, schema=self.schema):
+            bind.execute(CreateEnumType(self))
+
+    def drop(self, bind=None, checkfirst=True):
+        if not checkfirst or bind.dialect.has_type(bind, self.name, schema=self.schema):
+            bind.execute(DropEnumType(self))
+        
+    def _on_table_create(self, event, target, bind, **kw):
+        self.create(bind=bind, checkfirst=True)
+
+    def _on_metadata_create(self, event, target, bind, **kw):
+        if self.metadata is not None:
+            self.create(bind=bind, checkfirst=True)
+
+    def _on_metadata_drop(self, event, target, bind, **kw):
+        self.drop(bind=bind, checkfirst=True)
+
 colspecs = {
-    sqltypes.Interval:INTERVAL
+    sqltypes.Interval:INTERVAL,
+    sqltypes.Enum:ENUM,
 }
 
 ischema_names = {
@@ -309,6 +330,22 @@ class PGDDLCompiler(compiler.DDLCompiler):
     def visit_drop_sequence(self, drop):
         return "DROP SEQUENCE %s" % self.preparer.format_sequence(drop.element)
         
+    
+    def visit_create_enum_type(self, create):
+        type_ = create.element
+        
+        return "CREATE TYPE %s AS ENUM (%s)" % (
+            self.preparer.format_type(type_),
+            ",".join("'%s'" % e for e in type_.enums)
+        )
+
+    def visit_drop_enum_type(self, drop):
+        type_ = drop.element
+
+        return "DROP TYPE %s" % (
+            self.preparer.format_type(type_)
+        )
+        
     def visit_create_index(self, create):
         preparer = self.preparer
         index = create.element
@@ -361,6 +398,12 @@ class PGTypeCompiler(compiler.GenericTypeCompiler):
 
     def visit_datetime(self, type_):
         return self.visit_TIMESTAMP(type_)
+    
+    def visit_enum(self, type_):
+        return self.visit_ENUM(type_)
+        
+    def visit_ENUM(self, type_):
+        return self.dialect.identifier_preparer.format_type(type_)
         
     def visit_TIMESTAMP(self, type_):
         return "TIMESTAMP " + (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE"
@@ -396,6 +439,14 @@ class PGIdentifierPreparer(compiler.IdentifierPreparer):
             value = value[1:-1].replace(self.escape_to_quote, self.escape_quote)
         return value
 
+    def format_type(self, type_, use_schema=True):
+        if not type_.name:
+            raise exc.ArgumentError("Postgresql ENUM type requires a name.")
+        
+        name = self.quote(type_.name, type_.quote)
+        if not self.omit_schema and use_schema and type_.schema is not None:
+            name = self.quote_schema(type_.schema, type_.quote) + "." + name
+        return name
         
 class PGInspector(reflection.Inspector):
 
@@ -407,8 +458,12 @@ class PGInspector(reflection.Inspector):
 
         return self.dialect.get_table_oid(self.conn, table_name, schema,
                                           info_cache=self.info_cache)
-    
 
+class CreateEnumType(schema._CreateDropBase):
+  __visit_name__ = "create_enum_type"
+
+class DropEnumType(schema._CreateDropBase):
+  __visit_name__ = "drop_enum_type"
 
 class PGExecutionContext(default.DefaultExecutionContext):
     def fire_sequence(self, seq):
@@ -530,15 +585,19 @@ class PGDialect(default.DefaultDialect):
         if schema is None:
             cursor = connection.execute(
                 sql.text("select relname from pg_class c join pg_namespace n on "
-                    "n.oid=c.relnamespace where n.nspname=current_schema() and lower(relname)=:name",
-                    bindparams=[sql.bindparam('name', unicode(table_name.lower()), type_=sqltypes.Unicode)]
+                    "n.oid=c.relnamespace where n.nspname=current_schema() and "
+                    "lower(relname)=:name",
+                    bindparams=[
+                            sql.bindparam('name', unicode(table_name.lower()),
+                            type_=sqltypes.Unicode)]
                 )
             )
         else:
             cursor = connection.execute(
                 sql.text("select relname from pg_class c join pg_namespace n on "
                         "n.oid=c.relnamespace where n.nspname=:schema and lower(relname)=:name",
-                    bindparams=[sql.bindparam('name', unicode(table_name.lower()), type_=sqltypes.Unicode),
+                    bindparams=[
+                        sql.bindparam('name', unicode(table_name.lower()), type_=sqltypes.Unicode),
                         sql.bindparam('schema', unicode(schema), type_=sqltypes.Unicode)] 
                 )
             )
@@ -548,27 +607,62 @@ class PGDialect(default.DefaultDialect):
         if schema is None:
             cursor = connection.execute(
                         sql.text("SELECT relname FROM pg_class c join pg_namespace n on "
-                            "n.oid=c.relnamespace where relkind='S' and n.nspname=current_schema() and lower(relname)=:name",
-                            bindparams=[sql.bindparam('name', unicode(sequence_name.lower()), type_=sqltypes.Unicode)] 
+                            "n.oid=c.relnamespace where relkind='S' and n.nspname=current_schema()"
+                            " and lower(relname)=:name",
+                            bindparams=[
+                                sql.bindparam('name', unicode(sequence_name.lower()),
+                                type_=sqltypes.Unicode)
+                            ] 
                         )
                     )
         else:
             cursor = connection.execute(
-                        sql.text("SELECT relname FROM pg_class c join pg_namespace n on "
-                            "n.oid=c.relnamespace where relkind='S' and n.nspname=:schema and lower(relname)=:name",
-                            bindparams=[sql.bindparam('name', unicode(sequence_name.lower()), type_=sqltypes.Unicode),
-                                sql.bindparam('schema', unicode(schema), type_=sqltypes.Unicode)] 
-                        )
+                    sql.text("SELECT relname FROM pg_class c join pg_namespace n on "
+                        "n.oid=c.relnamespace where relkind='S' and n.nspname=:schema and "
+                        "lower(relname)=:name",
+                        bindparams=[
+                            sql.bindparam('name', unicode(sequence_name.lower()),
+                             type_=sqltypes.Unicode),
+                            sql.bindparam('schema', unicode(schema), type_=sqltypes.Unicode)
+                        ]
                     )
+                )
 
         return bool(cursor.first())
 
+    def has_type(self, connection, type_name, schema=None):
+        bindparams = [
+            sql.bindparam('typname',
+                unicode(type_name), type_=sqltypes.Unicode),
+            sql.bindparam('nspname',
+                unicode(schema), type_=sqltypes.Unicode),
+            ]
+        if schema is not None:
+            query = """
+            SELECT EXISTS (
+                SELECT * FROM pg_catalog.pg_type t, pg_catalog.pg_namespace n
+                WHERE t.typnamespace = n.oid
+                AND t.typname = :typname
+                AND n.nspname = :nspname
+                )
+                """
+        else:
+            query = """
+            SELECT EXISTS (
+                SELECT * FROM pg_catalog.pg_type t
+                WHERE t.typname = :typname
+                AND pg_type_is_visible(t.oid)
+                )
+                """
+        cursor = connection.execute(sql.text(query, bindparams=bindparams))
+        return bool(cursor.scalar())
+
     def table_names(self, connection, schema):
         result = connection.execute(
-            sql.text(u"""SELECT relname
-                FROM pg_class c
-                WHERE relkind = 'r'
-                AND '%s' = (select nspname from pg_namespace n where n.oid = c.relnamespace)""" % schema,
+            sql.text(u"SELECT relname FROM pg_class c "
+                "WHERE relkind = 'r' "
+                "AND '%s' = (select nspname from pg_namespace n where n.oid = c.relnamespace) " %
+                 schema,
                 typemap = {'relname':sqltypes.Unicode}
             )
         )
@@ -710,6 +804,8 @@ class PGDialect(default.DefaultDialect):
         c = connection.execute(s, table_oid=table_oid)
         rows = c.fetchall()
         domains = self._load_domains(connection)
+        enums = self._load_enums(connection)
+        
         # format columns
         columns = []
         for name, format_type, default, notnull, attnum, table_oid in rows:
@@ -748,18 +844,26 @@ class PGDialect(default.DefaultDialect):
                 kwargs['timezone'] = False
             if attype in self.ischema_names:
                 coltype = self.ischema_names[attype]
-            else:
-                if attype in domains:
-                    domain = domains[attype]
-                    if domain['attype'] in self.ischema_names:
-                        # A table can't override whether the domain is nullable.
-                        nullable = domain['nullable']
-                        if domain['default'] and not default:
-                            # It can, however, override the default value, but can't set it to null.
-                            default = domain['default']
-                        coltype = self.ischema_names[domain['attype']]
+            elif attype in enums:
+                enum = enums[attype]
+                coltype = ENUM
+                if "." in attype:
+                    kwargs['schema'], kwargs['name'] = attype.split('.')
                 else:
-                    coltype = None
+                    kwargs['name'] = attype
+                args = tuple(enum['labels'])
+            elif attype in domains:
+                domain = domains[attype]
+                if domain['attype'] in self.ischema_names:
+                    # A table can't override whether the domain is nullable.
+                    nullable = domain['nullable']
+                    if domain['default'] and not default:
+                        # It can, however, override the default value, but can't set it to null.
+                        default = domain['default']
+                    coltype = self.ischema_names[domain['attype']]
+            else:
+                coltype = None
+                
             if coltype:
                 coltype = coltype(*args, **kwargs)
                 if is_array:
@@ -883,6 +987,45 @@ class PGDialect(default.DefaultDialect):
             index_d['unique'] = unique
         return indexes
 
+    def _load_enums(self, connection):
+        ## Load data types for enums:
+        SQL_ENUMS = """
+            SELECT t.typname as "name",
+                   -- t.typdefault as "default", -- no enum defaults in 8.4 at least
+                   pg_catalog.pg_type_is_visible(t.oid) as "visible",
+                   n.nspname as "schema",
+                   e.enumlabel as "label"
+            FROM pg_catalog.pg_type t
+                 LEFT JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace
+                 LEFT JOIN pg_catalog.pg_constraint r ON t.oid = r.contypid
+                 LEFT JOIN pg_catalog.pg_enum e ON t.oid = e.enumtypid
+            WHERE t.typtype = 'e'
+            ORDER BY "name", e.oid -- e.oid gives us label order
+        """
+
+        s = sql.text(SQL_ENUMS, typemap={'attname':sqltypes.Unicode, 'label':sqltypes.Unicode})
+        c = connection.execute(s)
+
+        enums = {}
+        for enum in c.fetchall():
+            if enum['visible']:
+                # 'visible' just means whether or not the enum is in a
+                # schema that's on the search path -- or not overriden by
+                # a schema with higher presedence. If it's not visible,
+                # it will be prefixed with the schema-name when it's used.
+                name = enum['name']
+            else:
+                name = "%s.%s" % (enum['schema'], enum['name'])
+
+            if name in enums:
+                enums[name]['labels'].append(enum['label'])
+            else:
+                enums[name] = {
+                        'labels': [enum['label']],
+                        }
+
+        return enums
+
     def _load_domains(self, connection):
         ## Load data types for domains:
         SQL_DOMAINS = """
@@ -914,7 +1057,11 @@ class PGDialect(default.DefaultDialect):
             else:
                 name = "%s.%s" % (domain['schema'], domain['name'])
 
-            domains[name] = {'attype':attype, 'nullable': domain['nullable'], 'default': domain['default']}
+            domains[name] = {
+                    'attype':attype, 
+                    'nullable': domain['nullable'], 
+                    'default': domain['default']
+                }
 
         return domains
 
diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py
index 5b63f6e29..b99f79a8e 100644
--- a/lib/sqlalchemy/schema.py
+++ b/lib/sqlalchemy/schema.py
@@ -641,6 +641,9 @@ class Column(SchemaItem, expression.ColumnClause):
         self.foreign_keys = util.OrderedSet()
         self._table_events = set()
         
+        if isinstance(self.type, types.SchemaType):
+            self.type._set_parent(self)
+            
         if self.default is not None:
             if isinstance(self.default, (ColumnDefault, Sequence)):
                 args.append(self.default)
@@ -651,8 +654,10 @@ class Column(SchemaItem, expression.ColumnClause):
                 args.append(self.server_default)
             else:
                 args.append(DefaultClause(self.server_default))
+                
         if self.onupdate is not None:
             args.append(ColumnDefault(self.onupdate, for_update=True))
+            
         if self.server_onupdate is not None:
             if isinstance(self.server_onupdate, FetchedValue):
                 args.append(self.server_default)
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 5f5b31c68..c1b421843 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1246,6 +1246,9 @@ class GenericTypeCompiler(engine.TypeCompiler):
     def visit_unicode_text(self, type_): 
         return self.visit_TEXT(type_)
     
+    def visit_enum(self, type_):
+        raise NotImplementedError("Enum not supported generically")
+        
     def visit_null(self, type_):
         raise NotImplementedError("Can't generate DDL for the null type")
         
@@ -1264,7 +1267,8 @@ class IdentifierPreparer(object):
 
     illegal_initial_characters = ILLEGAL_INITIAL_CHARACTERS
 
-    def __init__(self, dialect, initial_quote='"', final_quote=None, escape_quote='"', omit_schema=False):
+    def __init__(self, dialect, initial_quote='"', 
+                    final_quote=None, escape_quote='"', omit_schema=False):
         """Construct a new ``IdentifierPreparer`` object.
 
         initial_quote
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 6a8a2c17e..b71c1892b 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -1187,43 +1187,32 @@ class ClauseElement(Visitable):
         """Compile this SQL expression.
 
         The return value is a :class:`~sqlalchemy.engine.Compiled` object.
-        Calling `str()` or `unicode()` on the returned value will yield a string
-        representation of the result. The :class:`~sqlalchemy.engine.Compiled`
-        object also can return a dictionary of bind parameter names and values
-        using the `params` accessor.
+        Calling ``str()`` or ``unicode()`` on the returned value will yield a
+        string representation of the result. The
+        :class:`~sqlalchemy.engine.Compiled` object also can return a
+        dictionary of bind parameter names and values
+        using the ``params`` accessor.
 
         :param bind: An ``Engine`` or ``Connection`` from which a
-          ``Compiled`` will be acquired.  This argument
-          takes precedence over this ``ClauseElement``'s
-          bound engine, if any.
+            ``Compiled`` will be acquired. This argument takes precedence over
+            this ``ClauseElement``'s bound engine, if any.
+
+        :param column_keys: Used for INSERT and UPDATE statements, a list of
+            column names which should be present in the VALUES clause of the
+            compiled statement. If ``None``, all columns from the target table
+            object are rendered.
 
         :param dialect: A ``Dialect`` instance frmo which a ``Compiled``
-          will be acquired.  This argument takes precedence
-          over the `bind` argument as well as this
-          ``ClauseElement``'s bound engine, if any.
+            will be acquired. This argument takes precedence over the `bind`
+            argument as well as this ``ClauseElement``'s bound engine, if any.
 
-        \**kw
-        
-          Keyword arguments are passed along to the compiler, 
-          which can affect the string produced.
-          
-          Keywords for a statement compiler are:
-        
-          column_keys
-            Used for INSERT and UPDATE statements, a list of
-            column names which should be present in the VALUES clause
-            of the compiled statement.  If ``None``, all columns
-            from the target table object are rendered.
-
-          inline
-            Used for INSERT statements, for a dialect which does
-            not support inline retrieval of newly generated
-            primary key columns, will force the expression used
-            to create the new primary key value to be rendered
-            inline within the INSERT statement's VALUES clause.
-            This typically refers to Sequence execution but
-            may also refer to any server-side default generation
-            function associated with a primary key `Column`.
+        :param inline: Used for INSERT statements, for a dialect which does
+            not support inline retrieval of newly generated primary key
+            columns, will force the expression used to create the new primary
+            key value to be rendered inline within the INSERT statement's
+            VALUES clause. This typically refers to Sequence execution but may
+            also refer to any server-side default generation function
+            associated with a primary key `Column`.
 
         """
         
@@ -3114,6 +3103,8 @@ class TableClause(_Immutable, FromClause):
             return []
 
     def count(self, whereclause=None, **params):
+        """return a SELECT COUNT generated against this ``TableClause``."""
+        
         if self.primary_key:
             col = list(self.primary_key)[0]
         else:
diff --git a/lib/sqlalchemy/types.py b/lib/sqlalchemy/types.py
index 9ae034c41..1fb7f2ed0 100644
--- a/lib/sqlalchemy/types.py
+++ b/lib/sqlalchemy/types.py
@@ -12,14 +12,13 @@ For more information see the SQLAlchemy documentation on types.
 
 """
 __all__ = [ 'TypeEngine', 'TypeDecorator', 'AbstractType', 'UserDefinedType',
-            'INT', 'CHAR', 'VARCHAR', 'NCHAR', 'NVARCHAR','TEXT', 'Text', 'FLOAT',
-            'NUMERIC', 'DECIMAL', 'TIMESTAMP', 'DATETIME', 'CLOB', 'BLOB',
-            'BOOLEAN', 'SMALLINT', 'INTEGER', 'DATE', 'TIME',
-            'String', 'Integer', 'SmallInteger', 'BigInteger', 
-            'Numeric', 'Float', 'DateTime', 'Date', 'Time', 'Binary',
-            'Boolean', 'Unicode', 'MutableType', 'Concatenable', 'UnicodeText', 'PickleType', 'Interval',
-            'type_map'
-            ]
+            'INT', 'CHAR', 'VARCHAR', 'NCHAR', 'NVARCHAR','TEXT', 'Text',
+            'FLOAT', 'NUMERIC', 'DECIMAL', 'TIMESTAMP', 'DATETIME', 'CLOB',
+            'BLOB', 'BOOLEAN', 'SMALLINT', 'INTEGER', 'DATE', 'TIME',
+            'String', 'Integer', 'SmallInteger', 'BigInteger', 'Numeric',
+            'Float', 'DateTime', 'Date', 'Time', 'Binary', 'Boolean',
+            'Unicode', 'MutableType', 'Concatenable', 'UnicodeText',
+            'PickleType', 'Interval', 'type_map', 'Enum' ]
 
 import inspect
 import datetime as dt
@@ -794,6 +793,147 @@ class Binary(TypeEngine):
     def get_dbapi_type(self, dbapi):
         return dbapi.BINARY
 
+class SchemaType(object):
+    """Mark a type as possibly requiring schema-level DDL for usage."""
+    
+    def __init__(self, **kw):
+        self.name = kw.pop('name', None)
+        self.quote = kw.pop('quote', None)
+        self.schema = kw.pop('schema', None)
+        self.metadata = kw.pop('metadata', None)
+        if self.metadata:
+            self.metadata.append_ddl_listener('before-create',
+                                                self._on_metadata_create)
+            self.metadata.append_ddl_listener('after-drop',
+                                                self._on_metadata_drop)
+            
+    def _set_parent(self, column):
+        column._on_table_attach(self._set_table)
+
+    def _set_table(self, table):
+        table.append_ddl_listener('before-create', self._on_table_create)
+        table.append_ddl_listener('after-drop', self._on_table_drop)
+        if self.metadata is None:
+            table.metadata.append_ddl_listener('before-create',
+                                                self._on_metadata_create)
+            table.metadata.append_ddl_listener('after-drop',
+                                                self._on_metadata_drop)
+
+    
+    @property
+    def bind(self):
+        return self.metadata and self.metadata.bind or None
+        
+    def create(self, bind=None, checkfirst=False):
+        from sqlalchemy.schema import _bind_or_error
+        if bind is None:
+            bind = _bind_or_error(self)
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t.create(bind=bind, checkfirst=checkfirst)
+
+    def drop(self, bind=None, checkfirst=False):
+        from sqlalchemy.schema import _bind_or_error
+        if bind is None:
+            bind = _bind_or_error(self)
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t.drop(bind=bind, checkfirst=checkfirst)
+        
+    def _on_table_create(self, event, target, bind, **kw):
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t._on_table_create(event, target, bind, **kw)
+
+    def _on_table_drop(self, event, target, bind, **kw):
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t._on_table_drop(event, target, bind, **kw)
+
+    def _on_metadata_create(self, event, target, bind, **kw):
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t._on_metadata_create(event, target, bind, **kw)
+
+    def _on_metadata_drop(self, event, target, bind, **kw):
+        t = self.dialect_impl(bind.dialect)
+        if t is not self:
+            t._on_metadata_drop(event, target, bind, **kw)
+    
+class Enum(String, SchemaType):
+    """Generic Enum Type.
+    
+    Currently supported on MySQL and Postgresql, the Enum type
+    provides a set of possible string values which the column is constrained
+    towards.
+    
+    Keyword arguments which don't apply to a specific backend are ignored
+    by that backend.
+    
+    :param \*enums: string or unicode enumeration labels. If unicode labels
+        are present, the `convert_unicode` flag is auto-enabled.
+    
+    :param assert_unicode: Enable unicode asserts for bind parameter values.
+        This flag is equivalent to that of ``String``.
+
+    :param convert_unicode: Enable unicode-aware bind parameter and result-set
+        processing for this Enum's data. This is set automatically based on
+        the presence of unicode label strings.
+
+    :param metadata: Associate this type directly with a ``MetaData`` object.
+        For types that exist on the target database as an independent schema
+        construct (Postgresql), this type will be created and dropped within
+        ``create_all()`` and ``drop_all()`` operations. If the type is not
+        associated with any ``MetaData`` object, it will associate itself with
+        each ``Table`` in which it is used, and will be created when any of
+        those individual tables are created, after a check is performed for
+        it's existence. The type is only dropped when ``drop_all()`` is called
+        for that ``Table`` object's metadata, however.
+    
+    :param name: The name of this type. This is required for Postgresql and
+        any future supported database which requires an explicitly named type,
+        or an explicitly named constraint in order to generate the type and/or
+        a table that uses it.
+    
+    :param schema: Schemaname of this type. For types that exist on the target
+        database as an independent schema construct (Postgresql), this
+        parameter specifies the named schema in which the type is present.
+    
+    :param quote: Force quoting to be on or off on the type's name. If left as
+        the default of `None`, the usual schema-level "case
+        sensitive"/"reserved name" rules are used to determine if this type's
+        name should be quoted.
+    
+    """
+    __visit_name__ = 'enum'
+    
+    def __init__(self, *enums, **kw):
+        self.enums = enums
+        convert_unicode= kw.pop('convert_unicode', None)
+        assert_unicode = kw.pop('assert_unicode', None)
+        if convert_unicode is None:
+            for e in enums:
+                if isinstance(e, unicode):
+                    convert_unicode = True
+                    break
+            else:
+                convert_unicode = False
+        
+        String.__init__(self, 
+                        convert_unicode=convert_unicode, 
+                        assert_unicode=assert_unicode
+                        )
+        SchemaType.__init__(self, **kw)
+        
+    def adapt(self, impltype):
+        return impltype(*self.enums, 
+                        name=self.name, 
+                        quote=self.quote, 
+                        schema=self.schema, 
+                        metadata=self.metadata,
+                        convert_unicode=self.convert_unicode,
+                        assert_unicode=self.assert_unicode
+                        )
 
 class PickleType(MutableType, TypeDecorator):
     """Holds Python objects.