the older "postgres_returning" and
"postgres_where" names still work with a
deprecation warning.
-
+
+ - Added new ENUM type, 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]
+
- using new dialect.initialize() feature to set up
version-dependent behavior.
- the "quoting='quoted'" argument to MSEnum/ENUM is deprecated.
It's best to rely upon the automatic quoting.
+ - ENUM now subclasses the new generic Enum type, and also handles
+ unicode values implicitly, if the given labelnames are unicode
+ objects.
+
- oracle
- unit tests pass 100% with cx_oracle !
the dialect itself if the type is not a standard SQL
type. This means reflection now returns more accurate
information about reflected types.
+
+ - 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]
- PickleType now uses == for comparison of values when
mutable=True, unless the "comparator" argument with a
.. autoclass:: Column
:members:
- :inherited-members:
:undoc-members:
:show-inheritance:
.. autoclass:: Table
:members:
- :inherited-members:
:undoc-members:
:show-inheritance:
type is emitted in ``CREATE TABLE``, such as ``VARCHAR`` see `SQL
Standard Types`_ and the other sections of this chapter.
-.. autoclass:: String
+.. autoclass:: Binary
:show-inheritance:
-.. autoclass:: Unicode
- :show-inheritance:
+.. autoclass:: Boolean
+ :show-inheritance:
-.. autoclass:: Text
- :show-inheritance:
+.. autoclass:: Date
+ :show-inheritance:
-.. autoclass:: UnicodeText
+.. autoclass:: DateTime
:show-inheritance:
+.. autoclass:: Enum
+ :show-inheritance:
+
+.. autoclass:: Float
+ :show-inheritance:
+
.. autoclass:: Integer
- :show-inheritance:
+ :show-inheritance:
-.. autoclass:: SmallInteger
- :show-inheritance:
+.. autoclass:: Interval
+ :show-inheritance:
.. autoclass:: Numeric
- :show-inheritance:
+ :show-inheritance:
-.. autoclass:: Float
- :show-inheritance:
+.. autoclass:: PickleType
+ :show-inheritance:
-.. autoclass:: DateTime
- :show-inheritance:
+.. autoclass:: SchemaType
+ :show-inheritance:
+ :members:
+ :undoc-members:
-.. autoclass:: Date
- :show-inheritance:
+.. autoclass:: SmallInteger
+ :show-inheritance:
-.. autoclass:: Time
+.. autoclass:: String
:show-inheritance:
-.. autoclass:: Interval
+.. autoclass:: Text
:show-inheritance:
-.. autoclass:: Boolean
- :show-inheritance:
+.. autoclass:: Time
+ :show-inheritance:
-.. autoclass:: Binary
- :show-inheritance:
+.. autoclass:: Unicode
+ :show-inheritance:
-.. autoclass:: PickleType
+.. autoclass:: UnicodeText
:show-inheritance:
-
SQL Standard Types
------------------
name when ``CREATE TABLE`` is issued. Some types may not be supported
on all databases.
-.. autoclass:: INT
- :show-inheritance:
+.. autoclass:: BLOB
+ :show-inheritance:
-.. autoclass:: sqlalchemy.types.INTEGER
- :show-inheritance:
+.. autoclass:: BOOLEAN
+ :show-inheritance:
.. autoclass:: CHAR
- :show-inheritance:
-
-.. autoclass:: VARCHAR
- :show-inheritance:
+ :show-inheritance:
-.. autoclass:: NCHAR
- :show-inheritance:
-
-.. autoclass:: TEXT
- :show-inheritance:
+.. autoclass:: CLOB
+ :show-inheritance:
-.. autoclass:: FLOAT
- :show-inheritance:
+.. autoclass:: DATE
+ :show-inheritance:
-.. autoclass:: NUMERIC
- :show-inheritance:
+.. autoclass:: DATETIME
+ :show-inheritance:
.. autoclass:: DECIMAL
- :show-inheritance:
+ :show-inheritance:
-.. autoclass:: TIMESTAMP
- :show-inheritance:
+.. autoclass:: FLOAT
+ :show-inheritance:
-.. autoclass:: DATETIME
- :show-inheritance:
+.. autoclass:: INT
+ :show-inheritance:
-.. autoclass:: CLOB
- :show-inheritance:
+.. autoclass:: sqlalchemy.types.INTEGER
+ :show-inheritance:
-.. autoclass:: BLOB
- :show-inheritance:
+.. autoclass:: NCHAR
+ :show-inheritance:
-.. autoclass:: BOOLEAN
- :show-inheritance:
+.. autoclass:: NUMERIC
+ :show-inheritance:
.. autoclass:: SMALLINT
- :show-inheritance:
+ :show-inheritance:
-.. autoclass:: DATE
- :show-inheritance:
+.. autoclass:: TEXT
+ :show-inheritance:
.. autoclass:: TIME
- :show-inheritance:
+ :show-inheritance:
+
+.. autoclass:: TIMESTAMP
+ :show-inheritance:
+
+.. autoclass:: VARCHAR
+ :show-inheritance:
Vendor-Specific Types
Column('col1', VARCHAR(200, collation='binary')),
Column('col2', TEXT(charset='latin1'))
)
-
-
Custom Types
------------
DECIMAL,
Date,
DateTime,
+ Enum,
FLOAT,
Float,
INT,
__visit_name__ = 'LONGBLOB'
-class ENUM(_StringType):
+class ENUM(sqltypes.Enum, _StringType):
"""MySQL ENUM type."""
__visit_name__ = 'ENUM'
'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):
sqltypes.Binary: _BinaryType,
sqltypes.Boolean: _MSBoolean,
sqltypes.Time: _MSTime,
+ sqltypes.Enum: ENUM,
}
# Everything 3.23 through 5.1 excepting OpenGIS types.
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
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)
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 = {
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
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"
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):
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):
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)]
)
)
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}
)
)
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:
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:
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 = """
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
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)
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)
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")
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
"""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`.
"""
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:
"""
__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
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.
-from sqlalchemy.test.testing import eq_
+# coding: utf-8
+
+from sqlalchemy.test.testing import eq_, assert_raises
# Py2K
import sets
from sqlalchemy.dialects.mysql import base as mysql
from sqlalchemy.test.testing import eq_
from sqlalchemy.test import *
+from sqlalchemy.test.engines import utf8_engine
class TypesTest(TestBase, AssertsExecutionResults):
Column('num4', mysql.MSDouble),
Column('num5', mysql.MSDouble()),
Column('num6', mysql.MSMediumInteger),
- Column('enum1', mysql.MSEnum("'black'", "'white'")),
- Column('enum2', mysql.MSEnum("dog", "cat")),
+ Column('enum1', mysql.ENUM("'black'", "'white'")),
+ Column('enum2', mysql.ENUM("dog", "cat")),
)
try:
table.drop(checkfirst=True)
assert isinstance(t2.c.num4.type, mysql.MSDouble)
assert isinstance(t2.c.num5.type, mysql.MSDouble)
assert isinstance(t2.c.num6.type, mysql.MSMediumInteger)
- assert isinstance(t2.c.enum1.type, mysql.MSEnum)
- assert isinstance(t2.c.enum2.type, mysql.MSEnum)
+ assert isinstance(t2.c.enum1.type, mysql.ENUM)
+ assert isinstance(t2.c.enum2.type, mysql.ENUM)
t2.drop()
t2.create()
finally:
(mysql.MSLongText, [], {'ascii':True},
'LONGTEXT ASCII'),
- (mysql.MSEnum, ["foo", "bar"], {'unicode':True},
+ (mysql.ENUM, ["foo", "bar"], {'unicode':True},
'''ENUM('foo','bar') UNICODE''')
]
db = testing.db
enum_table = Table('mysql_enum', MetaData(testing.db),
- Column('e1', mysql.MSEnum("'a'", "'b'")),
- Column('e2', mysql.MSEnum("'a'", "'b'"),
+ Column('e1', mysql.ENUM("'a'", "'b'")),
+ Column('e2', mysql.ENUM("'a'", "'b'"),
nullable=False),
- Column('e3', mysql.MSEnum("'a'", "'b'", strict=True)),
- Column('e4', mysql.MSEnum("'a'", "'b'", strict=True),
+ Column('e2generic', Enum("a", "b"),
+ nullable=False),
+ Column('e3', mysql.ENUM("'a'", "'b'", strict=True)),
+ Column('e4', mysql.ENUM("'a'", "'b'", strict=True),
nullable=False),
- Column('e5', mysql.MSEnum("a", "b")),
- Column('e6', mysql.MSEnum("'a'", "b")),
+ Column('e5', mysql.ENUM("a", "b")),
+ Column('e5generic', Enum("a", "b")),
+ Column('e6', mysql.ENUM("'a'", "b")),
)
eq_(colspec(enum_table.c.e1),
"e1 ENUM('a','b')")
eq_(colspec(enum_table.c.e2),
"e2 ENUM('a','b') NOT NULL")
+ eq_(colspec(enum_table.c.e2generic),
+ "e2generic ENUM('a','b') NOT NULL")
eq_(colspec(enum_table.c.e3),
"e3 ENUM('a','b')")
eq_(colspec(enum_table.c.e4),
"e4 ENUM('a','b') NOT NULL")
eq_(colspec(enum_table.c.e5),
"e5 ENUM('a','b')")
+ eq_(colspec(enum_table.c.e5generic),
+ "e5generic ENUM('a','b')")
eq_(colspec(enum_table.c.e6),
"e6 ENUM('''a''','b')")
enum_table.drop(checkfirst=True)
enum_table.create()
- try:
- enum_table.insert().execute(e1=None, e2=None, e3=None, e4=None)
- self.assert_(False)
- except exc.SQLError:
- self.assert_(True)
-
- try:
- enum_table.insert().execute(e1='c', e2='c', e3='c',
- e4='c', e5='c', e6='c')
- self.assert_(False)
- except exc.InvalidRequestError:
- self.assert_(True)
+ assert_raises(exc.SQLError, enum_table.insert().execute, e1=None, e2=None, e3=None, e4=None)
+
+ assert_raises(exc.InvalidRequestError, enum_table.insert().execute,
+ e1='c', e2='c', e2generic='c', e3='c',
+ e4='c', e5='c', e5generic='c', e6='c')
enum_table.insert().execute()
- enum_table.insert().execute(e1='a', e2='a', e3='a',
- e4='a', e5='a', e6="'a'")
- enum_table.insert().execute(e1='b', e2='b', e3='b',
- e4='b', e5='b', e6='b')
+ enum_table.insert().execute(e1='a', e2='a', e2generic='a', e3='a',
+ e4='a', e5='a', e5generic='a', e6="'a'")
+ enum_table.insert().execute(e1='b', e2='b', e2generic='b', e3='b',
+ e4='b', e5='b', e5generic='b', e6='b')
res = enum_table.select().execute().fetchall()
- expected = [(None, 'a', None, 'a', None, None),
- ('a', 'a', 'a', 'a', 'a', "'a'"),
- ('b', 'b', 'b', 'b', 'b', 'b')]
+ expected = [(None, 'a', 'a', None, 'a', None, None, None),
+ ('a', 'a', 'a', 'a', 'a', 'a', 'a', "'a'"),
+ ('b', 'b', 'b', 'b', 'b', 'b', 'b', 'b')]
# This is known to fail with MySQLDB 1.2.2 beta versions
# which return these as sets.Set(['a']), sets.Set(['b'])
eq_(res, expected)
enum_table.drop()
-
+
+ def test_unicode_enum(self):
+ unicode_engine = utf8_engine()
+ metadata = MetaData(unicode_engine)
+ t1 = Table('table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('value', Enum(u'réveillé', u'drôle', u'S’il')),
+ Column('value2', mysql.ENUM(u'réveillé', u'drôle', u'S’il'))
+ )
+ metadata.create_all()
+ try:
+ t1.insert().execute(value=u'drôle', value2=u'drôle')
+ t1.insert().execute(value=u'réveillé', value2=u'réveillé')
+ t1.insert().execute(value=u'S’il', value2=u'S’il')
+ eq_(t1.select().order_by(t1.c.id).execute().fetchall(),
+ [(1, u'drôle', u'drôle'), (2, u'réveillé', u'réveillé'), (3, u'S’il', u'S’il')]
+ )
+
+ # test reflection of the enum labels
+ m2 = MetaData(testing.db)
+ t2 = Table('table', m2, autoload=True)
+ # TODO: what's wrong with the last element ? is there
+ # latin-1 stuff forcing its way in ?
+ assert t2.c.value.type.enums[0:2] == (u'réveillé', u'drôle') #, u'S’il') # eh ?
+ assert t2.c.value2.type.enums[0:2] == (u'réveillé', u'drôle') #, u'S’il') # eh ?
+
+ finally:
+ metadata.drop_all()
+
+
@testing.exclude('mysql', '<', (4,), "3.23 can't handle an ENUM of ''")
@testing.uses_deprecated('Manually quoting ENUM value literals')
def test_enum_parse(self):
# MySQL 3.23 can't handle an ENUM of ''....
enum_table = Table('mysql_enum', MetaData(testing.db),
- Column('e1', mysql.MSEnum("'a'")),
- Column('e2', mysql.MSEnum("''")),
- Column('e3', mysql.MSEnum('a')),
- Column('e4', mysql.MSEnum('')),
- Column('e5', mysql.MSEnum("'a'", "''")),
- Column('e6', mysql.MSEnum("''", "'a'")),
- Column('e7', mysql.MSEnum("''", "'''a'''", "'b''b'", "''''")))
+ Column('e1', mysql.ENUM("'a'")),
+ Column('e2', mysql.ENUM("''")),
+ Column('e3', mysql.ENUM('a')),
+ Column('e4', mysql.ENUM('')),
+ Column('e5', mysql.ENUM("'a'", "''")),
+ Column('e6', mysql.ENUM("''", "'a'")),
+ Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
for col in enum_table.c:
self.assert_(repr(col))
reflected = Table('mysql_enum', MetaData(testing.db),
autoload=True)
for t in enum_table, reflected:
- eq_(t.c.e1.type.enums, ["a"])
- eq_(t.c.e2.type.enums, [""])
- eq_(t.c.e3.type.enums, ["a"])
- eq_(t.c.e4.type.enums, [""])
- eq_(t.c.e5.type.enums, ["a", ""])
- eq_(t.c.e6.type.enums, ["", "a"])
- eq_(t.c.e7.type.enums, ["", "'a'", "b'b", "'"])
+ eq_(t.c.e1.type.enums, ("a",))
+ eq_(t.c.e2.type.enums, ("",))
+ eq_(t.c.e3.type.enums, ("a",))
+ eq_(t.c.e4.type.enums, ("",))
+ eq_(t.c.e5.type.enums, ("a", ""))
+ eq_(t.c.e6.type.enums, ("", "a"))
+ eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
finally:
enum_table.drop()
( mysql.MSBlob(1234), mysql.MSBlob()),
( mysql.MSMediumBlob(),),
( mysql.MSLongBlob(),),
- ( mysql.MSEnum("''","'fleem'"), ),
+ ( mysql.ENUM("''","'fleem'"), ),
]
columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
+# coding: utf-8
from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message
from sqlalchemy.test import engines
import datetime
"SELECT EXTRACT(%s FROM t.col1::timestamp) AS anon_1 "
"FROM t" % field)
+class EnumTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL):
+ __only_on__ = 'postgresql'
+ __dialect__ = postgresql.dialect()
+
+ def test_compile(self):
+ e1 = Enum('x', 'y', 'z', name="somename")
+ e2 = Enum('x', 'y', 'z', name="somename", schema='someschema')
+
+ self.assert_compile(
+ postgresql.CreateEnumType(e1),
+ "CREATE TYPE somename AS ENUM ('x','y','z')"
+ )
+
+ self.assert_compile(
+ postgresql.CreateEnumType(e2),
+ "CREATE TYPE someschema.somename AS ENUM ('x','y','z')"
+ )
+
+ self.assert_compile(
+ postgresql.DropEnumType(e1),
+ "DROP TYPE somename"
+ )
+
+ self.assert_compile(
+ postgresql.DropEnumType(e2),
+ "DROP TYPE someschema.somename"
+ )
+
+ def test_create_table(self):
+ metadata = MetaData(testing.db)
+ t1 = Table('table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('value', Enum('one', 'two', 'three', name='onetwothreetype'))
+ )
+ t1.create()
+ t1.create(checkfirst=True) # check the create
+ try:
+ t1.insert().execute(value='two')
+ t1.insert().execute(value='three')
+ t1.insert().execute(value='three')
+ eq_(t1.select().order_by(t1.c.id).execute().fetchall(),
+ [(1, 'two'), (2, 'three'), (3, 'three')]
+ )
+ finally:
+ metadata.drop_all()
+ metadata.drop_all()
+
+ def test_name_required(self):
+ metadata = MetaData(testing.db)
+ etype = Enum('four', 'five', 'six', metadata=metadata)
+ assert_raises(exc.ArgumentError, etype.create)
+ assert_raises(exc.ArgumentError, etype.compile, dialect=postgresql.dialect())
+
+ def test_unicode_labels(self):
+ metadata = MetaData(testing.db)
+ t1 = Table('table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('value', Enum(u'réveillé', u'drôle', u'S’il', name='onetwothreetype'))
+ )
+ metadata.create_all()
+ try:
+ t1.insert().execute(value=u'drôle')
+ t1.insert().execute(value=u'réveillé')
+ t1.insert().execute(value=u'S’il')
+ eq_(t1.select().order_by(t1.c.id).execute().fetchall(),
+ [(1, u'drôle'), (2, u'réveillé'), (3, u'S’il')]
+ )
+
+ m2 = MetaData(testing.db)
+ t2 = Table('table', m2, autoload=True)
+ assert t2.c.value.type.enums == (u'réveillé', u'drôle', u'S’il')
+
+ finally:
+ metadata.drop_all()
+
+ def test_standalone_enum(self):
+ metadata = MetaData(testing.db)
+ etype = Enum('four', 'five', 'six', name='fourfivesixtype', metadata=metadata)
+ etype.create()
+ try:
+ assert testing.db.dialect.has_type(testing.db, 'fourfivesixtype')
+ finally:
+ etype.drop()
+ assert not testing.db.dialect.has_type(testing.db, 'fourfivesixtype')
+
+ metadata.create_all()
+ try:
+ assert testing.db.dialect.has_type(testing.db, 'fourfivesixtype')
+ finally:
+ metadata.drop_all()
+ assert not testing.db.dialect.has_type(testing.db, 'fourfivesixtype')
+
+ def test_reflection(self):
+ metadata = MetaData(testing.db)
+ etype = Enum('four', 'five', 'six', name='fourfivesixtype', metadata=metadata)
+ t1 = Table('table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('value', Enum('one', 'two', 'three', name='onetwothreetype')),
+ Column('value2', etype)
+ )
+ metadata.create_all()
+ try:
+ m2 = MetaData(testing.db)
+ t2 = Table('table', m2, autoload=True)
+ assert t2.c.value.type.enums == ('one', 'two', 'three')
+ assert t2.c.value.type.name == 'onetwothreetype'
+ assert t2.c.value2.type.enums == ('four', 'five', 'six')
+ assert t2.c.value2.type.name == 'fourfivesixtype'
+ finally:
+ metadata.drop_all()
+ def test_schema_reflection(self):
+ metadata = MetaData(testing.db)
+ etype = Enum('four', 'five', 'six',
+ name='fourfivesixtype',
+ schema='test_schema',
+ metadata=metadata)
+ t1 = Table('table', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('value', Enum('one', 'two', 'three',
+ name='onetwothreetype', schema='test_schema')),
+ Column('value2', etype)
+ )
+ metadata.create_all()
+ try:
+ m2 = MetaData(testing.db)
+ t2 = Table('table', m2, autoload=True)
+ assert t2.c.value.type.enums == ('one', 'two', 'three')
+ assert t2.c.value.type.name == 'onetwothreetype'
+ assert t2.c.value2.type.enums == ('four', 'five', 'six')
+ assert t2.c.value2.type.name == 'fourfivesixtype'
+ assert t2.c.value2.type.schema == 'test_schema'
+ finally:
+ metadata.drop_all()
+
+
+
+
class InsertTest(TestBase, AssertsExecutionResults):
__only_on__ = 'postgresql'