在Sqlalchemy做enum的最好方法是什么？

Question

我正在阅读有关sqlalchemy的内容,我看到以下代码:

employees_table = Table('employees', metadata,
    Column('employee_id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('manager_data', String(50)),
    Column('engineer_info', String(50)),
    Column('type', String(20), nullable=False)
)

employee_mapper = mapper(Employee, employees_table, \
    polymorphic_on=employees_table.c.type, polymorphic_identity='employee')
manager_mapper = mapper(Manager, inherits=employee_mapper, polymorphic_identity='manager')
engineer_mapper = mapper(Engineer, inherits=employee_mapper, polymorphic_identity='engineer')

我应该使用库中的常量使'type'成为一个int吗？或者我应该让make type为枚举？

Answer 1

从SQLAlchemy 1.1开始, Python的枚举类型可直接被SQLAlchemy Enum类型接受:

import enum
from sqlalchemy import Integer, Enum

class MyEnum(enum.Enum):
    one = 1
    two = 2
    three = 3

class MyClass(Base):
    __tablename__ = 'some_table'
    id = Column(Integer, primary_key=True)
    value = Column(Enum(MyEnum))

注意,在上面,字符串值"一","二","三"是持久的,而不是整数值.

对于旧版本的SQLAlchemy,我写了一篇文章,创建了自己的枚举类型(http://techspot.zzzeek.org/2011/01/14/the-enum-recipe/)

from sqlalchemy.types import SchemaType, TypeDecorator, Enum
from sqlalchemy import __version__
import re

if __version__ < '0.6.5':
    raise NotImplementedError("Version 0.6.5 or higher of SQLAlchemy is required.")

class EnumSymbol(object):
    """Define a fixed symbol tied to a parent class."""

    def __init__(self, cls_, name, value, description):
        self.cls_ = cls_
        self.name = name
        self.value = value
        self.description = description

    def __reduce__(self):
        """Allow unpickling to return the symbol 
        linked to the DeclEnum class."""
        return getattr, (self.cls_, self.name)

    def __iter__(self):
        return iter([self.value, self.description])

    def __repr__(self):
        return "<%s>" % self.name

class EnumMeta(type):
    """Generate new DeclEnum classes."""

    def __init__(cls, classname, bases, dict_):
        cls._reg = reg = cls._reg.copy()
        for k, v in dict_.items():
            if isinstance(v, tuple):
                sym = reg[v[0]] = EnumSymbol(cls, k, *v)
                setattr(cls, k, sym)
        return type.__init__(cls, classname, bases, dict_)

    def __iter__(cls):
        return iter(cls._reg.values())

class DeclEnum(object):
    """Declarative enumeration."""

    __metaclass__ = EnumMeta
    _reg = {}

    @classmethod
    def from_string(cls, value):
        try:
            return cls._reg[value]
        except KeyError:
            raise ValueError(
                    "Invalid value for %r: %r" % 
                    (cls.__name__, value)
                )

    @classmethod
    def values(cls):
        return cls._reg.keys()

    @classmethod
    def db_type(cls):
        return DeclEnumType(cls)

class DeclEnumType(SchemaType, TypeDecorator):
    def __init__(self, enum):
        self.enum = enum
        self.impl = Enum(
                        *enum.values(), 
                        name="ck%s" % re.sub(
                                    '([A-Z])', 
                                    lambda m:"_" + m.group(1).lower(), 
                                    enum.__name__)
                    )

    def _set_table(self, table, column):
        self.impl._set_table(table, column)

    def copy(self):
        return DeclEnumType(self.enum)

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return value.value

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return self.enum.from_string(value.strip())

if __name__ == '__main__':
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import Column, Integer, String, create_engine
    from sqlalchemy.orm import Session

    Base = declarative_base()

    class EmployeeType(DeclEnum):
        part_time = "P", "Part Time"
        full_time = "F", "Full Time"
        contractor = "C", "Contractor"

    class Employee(Base):
        __tablename__ = 'employee'

        id = Column(Integer, primary_key=True)
        name = Column(String(60), nullable=False)
        type = Column(EmployeeType.db_type())

        def __repr__(self):
             return "Employee(%r, %r)" % (self.name, self.type)

    e = create_engine('sqlite://', echo=True)
    Base.metadata.create_all(e)

    sess = Session(e)

    sess.add_all([
        Employee(name='e1', type=EmployeeType.full_time),
        Employee(name='e2', type=EmployeeType.full_time),
        Employee(name='e3', type=EmployeeType.part_time),
        Employee(name='e4', type=EmployeeType.contractor),
        Employee(name='e5', type=EmployeeType.contractor),
    ])
    sess.commit()

    print sess.query(Employee).filter_by(type=EmployeeType.contractor).all()

Answer 2

SQLAlchemy的具有自0.6枚举类型: http://docs.sqlalchemy.org/en/latest/core/type_basics.html?highlight=enum#sqlalchemy.types.Enum

虽然如果您的数据库具有本机枚举类型,我只建议使用它.否则我个人只会使用一个int.

Answer 3

我对SQLAlchemy并不是很了解,但是Paulo的这种方法对我来说似乎更简单.
我不需要用户友好的描述,所以我选择了它.

引用保罗(我希望他不介意我在此重新发布):

Python的namedtuple集合来拯救.顾名思义,a namedtuple是一个元组,每个项目都有一个名称.像普通的元组一样,项目是不可变的.与普通元组不同,可以使用点表示法通过名称访问项目的值.

这是一个用于创建的实用程序函数namedtuple:

from collections import namedtuple

def create_named_tuple(*values):
     return namedtuple('NamedTuple', values)(*values)

该*值之前变量是"拆包"的,使每个项目作为一个单独的参数传递给函数列表中的项目.

要创建一个namedtuple,只需使用所需的值调用上述函数:

>>> project_version = create_named_tuple('alpha', 'beta', 'prod')
NamedTuple(alpha='alpha', beta='beta', prod='prod')

我们现在可以使用project_versionnamedtuple来指定版本字段的值.

class Project(Base):
     ...
     version = Column(Enum(*project_version._asdict().values(), name='projects_version'))
     ...

这对我来说很有用,并且比我之前找到的其他解决方案简单得多.

Answer 4

注意:以下内容已过时.你现在应该按照Wolph的建议使用sqlalchemy.types.Enum.它特别好,因为它符合PEP-435自SQLAlchemy 1.1以来.

我喜欢zzzeek的食谱http://techspot.zzzeek.org/2011/01/14/the-enum-recipe/,但我改变了两件事:

• 我也使用EnumSymbol的Python名称作为数据库中的名称,而不是使用其值.我认为这不那么令人困惑.具有单独的值仍然是有用的,例如用于在UI中创建弹出菜单.该描述可以被认为是可以用于例如工具提示的值的更长版本.
• 在原始配方中,EnumSymbols的顺序是任意的,无论是在Python中迭代它们还是在数据库上执行"order by"时.但通常我想要一个确定的顺序.因此,如果将属性设置为字符串或元组,或者将属性显式设置为EnumSymbols时声明值的顺序,则将顺序更改为字母顺序 - 这是使用与SQLAlchemy命令列时相同的技巧在DeclarativeBase类中.

例子:

class EmployeeType(DeclEnum):
    # order will be alphabetic: contractor, part_time, full_time
    full_time = "Full Time"
    part_time = "Part Time"
    contractor = "Contractor"

class EmployeeType(DeclEnum):
    # order will be as stated: full_time, part_time, contractor
    full_time = EnumSymbol("Full Time")
    part_time = EnumSymbol("Part Time")
    contractor = EnumSymbol("Contractor")

这是修改后的食谱; 它使用Python 2.7中提供的OrderedDict类:

import re

from sqlalchemy.types import SchemaType, TypeDecorator, Enum
from sqlalchemy.util import set_creation_order, OrderedDict


class EnumSymbol(object):
    """Define a fixed symbol tied to a parent class."""

    def __init__(self, value, description=None):
        self.value = value
        self.description = description
        set_creation_order(self)

    def bind(self, cls, name):
        """Bind symbol to a parent class."""
        self.cls = cls
        self.name = name
        setattr(cls, name, self)

    def __reduce__(self):
        """Allow unpickling to return the symbol linked to the DeclEnum class."""
        return getattr, (self.cls, self.name)

    def __iter__(self):
        return iter([self.value, self.description])

    def __repr__(self):
        return "<%s>" % self.name


class DeclEnumMeta(type):
    """Generate new DeclEnum classes."""

    def __init__(cls, classname, bases, dict_):
        reg = cls._reg = cls._reg.copy()
        for k in sorted(dict_):
            if k.startswith('__'):
                continue
            v = dict_[k]
            if isinstance(v, basestring):
                v = EnumSymbol(v)
            elif isinstance(v, tuple) and len(v) == 2:
                v = EnumSymbol(*v)
            if isinstance(v, EnumSymbol):
                v.bind(cls, k)
                reg[k] = v
        reg.sort(key=lambda k: reg[k]._creation_order)
        return type.__init__(cls, classname, bases, dict_)

    def __iter__(cls):
        return iter(cls._reg.values())


class DeclEnum(object):
    """Declarative enumeration.

    Attributes can be strings (used as values),
    or tuples (used as value, description) or EnumSymbols.
    If strings or tuples are used, order will be alphabetic,
    otherwise order will be as in the declaration.

    """

    __metaclass__ = DeclEnumMeta
    _reg = OrderedDict()

    @classmethod
    def names(cls):
        return cls._reg.keys()

    @classmethod
    def db_type(cls):
        return DeclEnumType(cls)


class DeclEnumType(SchemaType, TypeDecorator):
    """DeclEnum augmented so that it can persist to the database."""

    def __init__(self, enum):
        self.enum = enum
        self.impl = Enum(*enum.names(), name="ck%s" % re.sub(
            '([A-Z])', lambda m: '_' + m.group(1).lower(), enum.__name__))

    def _set_table(self, table, column):
        self.impl._set_table(table, column)

    def copy(self):
        return DeclEnumType(self.enum)

    def process_bind_param(self, value, dialect):
        if isinstance(value, EnumSymbol):
            value = value.name
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            return getattr(self.enum, value.strip())

Answer 5

这个和相关的 StackOverflow 线程答案诉诸于 PostgreSQL 或其他特定于方言的类型。然而，在 SQLAlchemy 中可以轻松实现通用支持，并且与 Alembic 迁移兼容。

如果后端不支持 Enum，SQLAlchemy 和 alembic 可以促进对 varchar 和类似类型的约束，以模拟枚举列类型。

首先，在要声明自定义 SQLAlchemy Enum 列类型的位置导入 Python 枚举、SQLAlchemy 枚举和SQLAlchemy 声明基。

import enum
from sqlalchemy import Enum
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

我们以OP原来的Python枚举类为例：

class PostStatus(enum.Enum):
    DRAFT='draft'
    APPROVE='approve'
    PUBLISHED='published'

现在我们创建一个SQLAlchemy Enum实例化：

PostStatusType: Enum = Enum(
    PostStatus,
    name="post_status_type",
    create_constraint=True,
    metadata=Base.metadata,
    validate_strings=True,
)

当您运行 Alembicalembic revision --autogenerate -m "Revision Notes"并尝试使用 来应用修订版时alembic upgrade head，您可能会收到有关类型不存在的错误。例如：

...
sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedObject) type "post_status_type" does not exist
LINE 10:  post_status post_status_type NOT NULL,
...

要解决此问题，请导入 SQLAlchemy Enum 类并将以下内容添加到Alembic 自动生成的修订脚本中的upgrade()和函数中。downgrade()

from myproject.database import PostStatusType
...
def upgrade() -> None:
    PostStatusType.create(op.get_bind(), checkfirst=True)
    ... the remainder of the autogen code...
def downgrade() -> None:
    ...the autogen code...
    PostStatusType.drop(op.get_bind(), checkfirst=True)

最后，请确保sa.Column()使用枚举类型更新表中自动生成的声明，以简单地引用 SQLAlchemy 枚举类型，而不是使用 Alembic 尝试重新声明它。例如在def upgrade() -> None:

op.create_table(
    "my_table",
    sa.Column(
        "post_status",
        PostStatusType,
        nullable=False,
    ),
)

Answer 6

对于 mysql 我使用它的方言

from sqlalchemy.dialects.mysql import ENUM

... 

class Videos(Base):
    ...
    video_type  = Column(ENUM('youtube', 'vimeo'))
    ...