.. changelog::
:version: 1.1.0b1
+ .. change::
+ :tags: feature, sql
+ :tickets: 3292, 3095
+
+ Added support for PEP-435-style enumerated classes, namely
+ Python 3's ``enum.Enum`` class but also including compatible
+ enumeration libraries, to the :class:`.types.Enum` datatype.
+ The :class:`.types.Enum` datatype now also performs in-Python validation
+ of incoming values, and adds an option to forego creating the
+ CHECK constraint :paramref:`.Enum.create_constraint`.
+ Pull request courtesy Alex Grönholm.
+
+ .. seealso::
+
+ :ref:`change_3292`
+
+ :ref:`change_3095`
+
.. change::
:tags: change, postgresql
:ticket:`3501`
+.. _change_3292:
+
+Support for Python's native ``enum`` type and compatible forms
+---------------------------------------------------------------
+
+The :class:`.Enum` type can now be constructed using any
+PEP-435 compliant enumerated type. When using this mode, input values
+and return values are the actual enumerated objects, not the
+string values::
+
+ import enum
+ from sqlalchemy import Table, MetaData, Column, Enum, create_engine
+
+
+ class MyEnum(enum.Enum):
+ one = "one"
+ two = "two"
+ three = "three"
+
+
+ t = Table(
+ 'data', MetaData(),
+ Column('value', Enum(MyEnum))
+ )
+
+ e = create_engine("sqlite://")
+ t.create(e)
+
+ e.execute(t.insert(), {"value": MyEnum.two})
+ assert e.scalar(t.select()) is MyEnum.two
+
+
+:ticket:`3292`
+
+.. _change_3095:
+
+The ``Enum`` type now does in-Python validation of values
+---------------------------------------------------------
+
+To accomodate for Python native enumerated objects, as well as for edge
+cases such as that of where a non-native ENUM type is used within an ARRAY
+and a CHECK contraint is infeasible, the :class:`.Enum` datatype now adds
+in-Python validation of input values::
+
+
+ >>> from sqlalchemy import Table, MetaData, Column, Enum, create_engine
+ >>> t = Table(
+ ... 'data', MetaData(),
+ ... Column('value', Enum("one", "two", "three"))
+ ... )
+ >>> e = create_engine("sqlite://")
+ >>> t.create(e)
+ >>> e.execute(t.insert(), {"value": "four"})
+ Traceback (most recent call last):
+ ...
+ sqlalchemy.exc.StatementError: (exceptions.LookupError)
+ "four" is not among the defined enum values
+ [SQL: u'INSERT INTO data (value) VALUES (?)']
+ [parameters: [{'value': 'four'}]]
+
+For simplicity and consistency, this validation is now turned on in all cases,
+whether or not the enumerated type uses a database-native form, whether
+or not the CHECK constraint is in use, as well as whether or not a
+PEP-435 enumerated type or plain list of string values is used. The
+check also occurs on the result-handling side as well, when values coming
+from the database are returned.
+
+This validation is in addition to the existing behavior of creating a
+CHECK constraint when a non-native enumerated type is used. The creation of
+this CHECK constraint can now be disabled using the new
+:paramref:`.Enum.create_constraint` flag.
+
+:ticket:`3095`
+
.. _change_2528:
A UNION or similar of SELECTs with LIMIT/OFFSET/ORDER BY now parenthesizes the embedded selects
:param enums: The range of valid values for this ENUM. Values will be
quoted when generating the schema according to the quoting flag (see
- below).
+ below). This object may also be a PEP-435-compliant enumerated
+ type.
- :param strict: Defaults to False: ensure that a given value is in this
- ENUM's range of permissible values when inserting or updating rows.
- Note that MySQL will not raise a fatal error if you attempt to store
- an out of range value- an alternate value will be stored instead.
- (See MySQL ENUM documentation.)
+ .. versionadded: 1.1 added support for PEP-435-compliant enumerated
+ types.
+
+ :param strict: This flag has no effect.
+
+ .. versionchanged:: The MySQL ENUM type as well as the base Enum
+ type now validates all Python data values.
:param charset: Optional, a column-level character set for this string
value. Takes precedence to 'ascii' or 'unicode' short-hand.
literals for you. This is a transitional option.
"""
- values, length = self._init_values(enums, kw)
- self.strict = kw.pop('strict', False)
+
+ kw.pop('strict', None)
+ sqltypes.Enum.__init__(self, *enums)
kw.pop('metadata', None)
kw.pop('schema', None)
kw.pop('name', 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)
+ _StringType.__init__(self, length=self.length, **kw)
+
+ def _setup_for_values(self, values, objects, kw):
+ values, length = self._init_values(values, kw)
+ return sqltypes.Enum._setup_for_values(self, values, objects, kw)
def __repr__(self):
return util.generic_repr(
self, to_inspect=[ENUM, _StringType, sqltypes.Enum])
- def bind_processor(self, dialect):
- super_convert = super(ENUM, self).bind_processor(dialect)
-
- def process(value):
- if self.strict and value is not None and value not in self.enums:
- raise exc.InvalidRequestError('"%s" not a valid value for '
- 'this enum' % value)
- if super_convert:
- return super_convert(value)
- else:
- return value
- return process
-
def adapt(self, cls, **kw):
- if issubclass(cls, ENUM):
- kw['strict'] = self.strict
return sqltypes.Enum.adapt(self, cls, **kw)
"""Generic Enum Type.
- The Enum type provides a set of possible string values which the
- column is constrained towards.
+ The :class:`.Enum` type provides a set of possible string values
+ which the column is constrained towards.
+
+ The :class:`.Enum` type will make use of the backend's native "ENUM"
+ type if one is available; otherwise, it uses a VARCHAR datatype and
+ produces a CHECK constraint. Use of the backend-native enum type
+ can be disabled using the :paramref:`.Enum.native_enum` flag, and
+ the production of the CHECK constraint is configurable using the
+ :paramref:`.Enum.create_constraint` flag.
+
+ The :class:`.Enum` type also provides in-Python validation of both
+ input values and database-returned values. A ``LookupError`` is raised
+ for any Python value that's not located in the given list of possible
+ values.
+
+ .. versionchanged:: 1.1 the :class:`.Enum` type now provides in-Python
+ validation of input values as well as on data being returned by
+ the database.
+
+ The source of enumerated values may be a list of string values, or
+ alternatively a PEP-435-compliant enumerated class. For the purposes
+ of the :class:`.Enum` datatype, this class need only provide a
+ ``__members__`` method.
+
+ When using an enumerated class, the enumerated objects are used
+ both for input and output, rather than strings as is the case with
+ a plain-string enumerated type::
+
+ import enum
+ class MyEnum(enum.Enum):
+ one = "one"
+ two = "two"
+ three = "three"
+
+
+ t = Table(
+ 'data', MetaData(),
+ Column('value', Enum(MyEnum))
+ )
+
+ connection.execute(t.insert(), {"value": MyEnum.two})
+ assert connection.scalar(t.select()) is MyEnum.two
+
+ .. versionadded:: 1.1 - support for PEP-435-style enumerated
+ classes.
- By default, uses the backend's native ENUM type if available,
- else uses VARCHAR + a CHECK constraint.
.. seealso::
Keyword arguments which don't apply to a specific backend are ignored
by that backend.
- :param \*enums: either exactly one PEP 435 compliant enumerated type
+ :param \*enums: either exactly one PEP-435 compliant enumerated type
or one or more string or unicode enumeration labels. If unicode
labels are present, the `convert_unicode` flag is auto-enabled.
+ .. versionadded:: 1.1 a PEP-435 style enumerated class may be
+ passed.
+
: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 create_constraint: defaults to True. When creating a non-native
+ enumerated type, also build a CHECK constraint on the database
+ against the valid values.
+
+ .. versionadded:: 1.1 - added :paramref:`.Enum.create_constraint`
+ which provides the option to disable the production of the
+ CHECK constraint for a non-native enumerated type.
+
: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
: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. If an :class:`~enum.Enum`
+ the type and/or a table that uses it. If a PEP-435 enumerated
class was used, its name (converted to lower case) is used by
default.
``schema`` attribute. This also takes effect when using the
:meth:`.Table.tometadata` operation.
- .. versionadded:: 0.8
-
"""
- if len(enums) == 1 and hasattr(enums[0], '__members__'):
- self.enums = list(enums[0].__members__)
- self.enum_class = enums[0]
- kw.setdefault('name', enums[0].__name__.lower())
- self.key_lookup = dict((value, key) for key, value in enums[0].__members__.items())
- self.value_lookup = enums[0].__members__.copy()
- else:
- self.enums = enums
- self.enum_class = self.key_lookup = self.value_lookup = None
+
+ values, objects = self._parse_into_values(enums, kw)
+ self._setup_for_values(values, objects, kw)
self.native_enum = kw.pop('native_enum', True)
convert_unicode = kw.pop('convert_unicode', None)
+ self.create_constraint = kw.pop('create_constraint', True)
if convert_unicode is None:
for e in self.enums:
if isinstance(e, util.text_type):
length = max(len(x) for x in self.enums)
else:
length = 0
+ self._valid_lookup[None] = self._object_lookup[None] = None
+
String.__init__(self,
length=length,
convert_unicode=convert_unicode,
)
SchemaType.__init__(self, **kw)
+ def _parse_into_values(self, enums, kw):
+ if len(enums) == 1 and hasattr(enums[0], '__members__'):
+ self.enum_class = enums[0]
+ values = list(self.enum_class.__members__)
+ objects = [self.enum_class.__members__[k] for k in values]
+ kw.setdefault('name', self.enum_class.__name__.lower())
+
+ return values, objects
+ else:
+ self.enum_class = None
+ return enums, enums
+
+ def _setup_for_values(self, values, objects, kw):
+ self.enums = list(values)
+
+ self._valid_lookup = dict(
+ zip(objects, values)
+ )
+ self._object_lookup = dict(
+ (value, key) for key, value in self._valid_lookup.items()
+ )
+ self._valid_lookup.update(
+ [(value, value) for value in self._valid_lookup.values()]
+ )
+
+ def _db_value_for_elem(self, elem):
+ try:
+ return self._valid_lookup[elem]
+ except KeyError:
+ raise LookupError(
+ '"%s" is not among the defined enum values' % elem)
+
+ def _object_value_for_elem(self, elem):
+ try:
+ return self._object_lookup[elem]
+ except KeyError:
+ raise LookupError(
+ '"%s" is not among the defined enum values' % elem)
+
def __repr__(self):
return util.generic_repr(self,
additional_kw=[('native_enum', True)],
if self.native_enum:
SchemaType._set_table(self, column, table)
+ if not self.create_constraint:
+ return
+
e = schema.CheckConstraint(
type_coerce(column, self).in_(self.enums),
name=_defer_name(self.name),
metadata = kw.pop('metadata', self.metadata)
_create_events = kw.pop('_create_events', False)
if issubclass(impltype, Enum):
- args = [self.enum_class] if self.enum_class is not None else self.enums
+ if self.enum_class is not None:
+ args = [self.enum_class]
+ else:
+ args = self.enums
return impltype(name=self.name,
schema=schema,
metadata=metadata,
def literal_processor(self, dialect):
parent_processor = super(Enum, self).literal_processor(dialect)
- if self.key_lookup:
- def process(value):
- value = self.key_lookup.get(value, value)
- if parent_processor:
- return parent_processor(value)
- return process
- else:
- return parent_processor
+ def process(value):
+ value = self._db_value_for_elem(value)
+ if parent_processor:
+ value = parent_processor(value)
+ return value
+ return process
def bind_processor(self, dialect):
def process(value):
- if isinstance(value, util.string_types):
- if value not in self.enums:
- raise LookupError(
- '"%s" is not among the defined enum values' %
- value)
- elif self.key_lookup and value in self.key_lookup:
- value = self.key_lookup[value]
-
+ value = self._db_value_for_elem(value)
if parent_processor:
value = parent_processor(value)
return value
return process
def result_processor(self, dialect, coltype):
- parent_processor = super(Enum, self).result_processor(dialect,
- coltype)
- if self.value_lookup:
- def process(value):
- if parent_processor:
- value = parent_processor(value)
+ parent_processor = super(Enum, self).result_processor(
+ dialect, coltype)
- try:
- return self.value_lookup[value]
- except KeyError:
- raise LookupError('No such member in enum class %s: %s' %
- (self.enum_class.__name__, value))
+ def process(value):
+ if parent_processor:
+ value = parent_processor(value)
- return process
- else:
- return parent_processor
+ value = self._object_value_for_elem(value)
+ return value
+
+ return process
@property
def python_type(self):
class PickleType(TypeDecorator):
-
"""Holds Python objects, which are serialized using pickle.
PickleType builds upon the Binary type to apply Python's
eq_(
t2.c.value.type.enums[0:2],
- (u('réveillé'), u('drôle')) # u'S’il') # eh ?
+ [u('réveillé'), u('drôle')] # u'S’il') # eh ?
)
eq_(
t2.c.value2.type.enums[0:2],
- (u('réveillé'), u('drôle')) # u'S’il') # eh ?
+ [u('réveillé'), u('drôle')] # u'S’il') # eh ?
)
def test_enum_compile(self):
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", "'"])
@testing.provide_metadata
@testing.exclude('mysql', '<', (5,))
from sqlalchemy.testing import mock
-class SomeEnum(object):
- # Implements PEP 435 in the minimal fashion needed by SQLAlchemy
- __members__ = OrderedDict()
-
- def __init__(self, name, value):
- self.name = name
- self.value = value
- self.__members__[name] = self
- setattr(SomeEnum, name, self)
-
-SomeEnum('one', 1)
-SomeEnum('two', 2)
-SomeEnum('three', 3)
class AdaptTest(fixtures.TestBase):
eq_(types.Unicode().python_type, util.text_type)
eq_(types.String(convert_unicode=True).python_type, util.text_type)
eq_(types.Enum('one', 'two', 'three').python_type, str)
- eq_(types.Enum(SomeEnum).python_type, SomeEnum)
assert_raises(
NotImplementedError,
Column('Lar', LargeBinary()),
Column('Pic', PickleType()),
Column('Int', Interval()),
- Column('Enu', Enum('x', 'y', 'z', name="somename")),
- Column('En2', Enum(SomeEnum)),
]
for column_type in column_types:
meta = MetaData()
class EnumTest(AssertsCompiledSQL, fixtures.TablesTest):
+ __backend__ = True
+
+ class SomeEnum(object):
+ # Implements PEP 435 in the minimal fashion needed by SQLAlchemy
+ __members__ = OrderedDict()
+
+ def __init__(self, name, value):
+ self.name = name
+ self.value = value
+ self.__members__[name] = self
+ setattr(self.__class__, name, self)
+
+ one = SomeEnum('one', 1)
+ two = SomeEnum('two', 2)
+ three = SomeEnum('three', 3)
@classmethod
def define_tables(cls, metadata):
'non_native_enum_table', metadata,
Column("id", Integer, primary_key=True),
Column('someenum', Enum('one', 'two', 'three', native_enum=False)),
+ Column('someotherenum',
+ Enum('one', 'two', 'three',
+ create_constraint=False, native_enum=False)),
)
Table(
'stdlib_enum_table', metadata,
Column("id", Integer, primary_key=True),
- Column('someenum', Enum(SomeEnum))
+ Column('someenum', Enum(cls.SomeEnum))
+ )
+
+ def test_python_type(self):
+ eq_(types.Enum(self.SomeEnum).python_type, self.SomeEnum)
+
+ def test_pickle_types(self):
+ global SomeEnum
+ SomeEnum = self.SomeEnum
+ for loads, dumps in picklers():
+ column_types = [
+ Column('Enu', Enum('x', 'y', 'z', name="somename")),
+ Column('En2', Enum(self.SomeEnum)),
+ ]
+ for column_type in column_types:
+ meta = MetaData()
+ Table('foo', meta, column_type)
+ loads(dumps(column_type))
+ loads(dumps(meta))
+
+ def test_validators_pep435(self):
+ type_ = Enum(self.SomeEnum)
+
+ bind_processor = type_.bind_processor(testing.db.dialect)
+ eq_(bind_processor('one'), "one")
+ eq_(bind_processor(self.one), "one")
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ bind_processor, "foo"
+ )
+
+ result_processor = type_.result_processor(testing.db.dialect, None)
+
+ eq_(result_processor('one'), self.one)
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ result_processor, "foo"
)
+ literal_processor = type_.literal_processor(testing.db.dialect)
+ eq_(literal_processor("one"), "'one'")
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ literal_processor, "foo"
+ )
+
+ def test_validators_plain(self):
+ type_ = Enum("one", "two")
+
+ bind_processor = type_.bind_processor(testing.db.dialect)
+ eq_(bind_processor('one'), "one")
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ bind_processor, "foo"
+ )
+
+ result_processor = type_.result_processor(testing.db.dialect, None)
+
+ eq_(result_processor('one'), "one")
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ result_processor, "foo"
+ )
+
+ literal_processor = type_.literal_processor(testing.db.dialect)
+ eq_(literal_processor("one"), "'one'")
+ assert_raises_message(
+ LookupError,
+ '"foo" is not among the defined enum values',
+ literal_processor, "foo"
+ )
+
+
@testing.fails_on(
'postgresql+zxjdbc',
'zxjdbc fails on ENUM: column "XXX" is of type XXX '
]
)
+ def test_null_round_trip(self):
+ enum_table = self.tables.enum_table
+ non_native_enum_table = self.tables.non_native_enum_table
+
+ with testing.db.connect() as conn:
+ conn.execute(enum_table.insert(), {"id": 1, "someenum": None})
+ eq_(conn.scalar(select([enum_table.c.someenum])), None)
+
+ with testing.db.connect() as conn:
+ conn.execute(
+ non_native_enum_table.insert(), {"id": 1, "someenum": None})
+ eq_(conn.scalar(select([non_native_enum_table.c.someenum])), None)
+
+
+ @testing.fails_on(
+ 'mysql',
+ "The CHECK clause is parsed but ignored by all storage engines.")
+ @testing.fails_on(
+ 'mssql', "FIXME: MS-SQL 2005 doesn't honor CHECK ?!?")
+ def test_check_constraint(self):
+ assert_raises(
+ (exc.IntegrityError, exc.ProgrammingError),
+ testing.db.execute,
+ "insert into non_native_enum_table "
+ "(id, someenum) values(1, 'four')")
+
+ def test_skip_check_constraint(self):
+ with testing.db.connect() as conn:
+ conn.execute(
+ "insert into non_native_enum_table "
+ "(id, someotherenum) values(1, 'four')"
+ )
+ eq_(
+ conn.scalar("select someotherenum from non_native_enum_table"),
+ "four")
+ assert_raises_message(
+ LookupError,
+ '"four" is not among the defined enum values',
+ conn.scalar,
+ select([self.tables.non_native_enum_table.c.someotherenum])
+ )
+
def test_non_native_round_trip(self):
non_native_enum_table = self.tables['non_native_enum_table']
])
eq_(
- non_native_enum_table.select().
+ select([
+ non_native_enum_table.c.id,
+ non_native_enum_table.c.someenum]).
order_by(non_native_enum_table.c.id).execute().fetchall(),
[
(1, 'two'),
]
)
- def test_stdlib_enum_round_trip(self):
+ def test_pep435_enum_round_trip(self):
stdlib_enum_table = self.tables['stdlib_enum_table']
stdlib_enum_table.insert().execute([
- {'id': 1, 'someenum': SomeEnum.two},
- {'id': 2, 'someenum': SomeEnum.two},
- {'id': 3, 'someenum': SomeEnum.one},
+ {'id': 1, 'someenum': self.SomeEnum.two},
+ {'id': 2, 'someenum': self.SomeEnum.two},
+ {'id': 3, 'someenum': self.SomeEnum.one},
])
eq_(
stdlib_enum_table.select().
order_by(stdlib_enum_table.c.id).execute().fetchall(),
[
- (1, SomeEnum.two),
- (2, SomeEnum.two),
- (3, SomeEnum.one),
+ (1, self.SomeEnum.two),
+ (2, self.SomeEnum.two),
+ (3, self.SomeEnum.one),
]
)
e1 = Enum('one', 'two', 'three', name='foo', schema='bar')
eq_(e1.adapt(ENUM).name, 'foo')
eq_(e1.adapt(ENUM).schema, 'bar')
- e1 = Enum(SomeEnum)
+ e1 = Enum(self.SomeEnum)
eq_(e1.adapt(ENUM).name, 'someenum')
eq_(e1.adapt(ENUM).enums, ['one', 'two', 'three'])
def test_lookup_failure(self):
assert_raises(
- exc.StatementError, self.tables['non_native_enum_table'].insert().execute,
+ exc.StatementError,
+ self.tables['non_native_enum_table'].insert().execute,
{'id': 4, 'someenum': 'four'}
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
