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

我正在阅读 sql 炼金术，我看到了以下代码:

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')

我是否应该在库中使用常量“键入”一个 int？或者我应该只使类型为枚举？

134320 次浏览
小开
最佳答案

SQLAlchemy 从0.6开始就有一个 Enum 类型: Http://docs.sqlalchemy.org/en/latest/core/type_basics.html?highlight=Enum#sqlalchemy.types

不过，我建议只有在数据库具有本机枚举类型时才使用它。否则我个人会使用 int。

小开

从 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))

注意，上面的字符串值“ one”、“ two”、“ three”是持久化的，而不是整数值。

For older versions of SQLAlchemy, I wrote a post which creates its own Enumerated type (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()
小开

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

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

  • 我在数据库中也使用了 Python 名称 Enum 符号作为名称，而不是使用它的值。我觉得这样就不那么让人困惑了。有一个单独的值仍然是有用的，例如在 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中可用的 OrderedDect 类:

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())
小开

我对 SQLAlchemy 并不是很了解，但是 保罗的方法对我来说似乎简单得多。
我不需要用户友好的描述，所以我就顺其自然了。

引用 Paulo (我希望他不介意我在这里转发) :

Python 的 namedtuple收藏来拯救。顾名思义，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_version namedtuple 来指定 version 字段的值。

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

这对我来说非常有效，而且比我之前找到的其他解决方案要简单得多。

小开

对于 mysql，我使用它的方言

from sqlalchemy.dialects.mysql import ENUM


...


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

这是我使用的方法——使用 IntEnum

from enum import IntEnum
class GenderType(IntEnum):
FEMALE:         int = 1
MALE:           int = 2
TRANSGENDER:    int = 3


class Citizen(Base):
__tablename__ = 'citizen'


user_uuid:      int = Column(UUID(as_uuid=True), primary_key=True)
gender_type:    int = Column(Integer, nullable=False, default=GenderType.MALE)
full_name:      str = Column(String(64))
address:        str = Column(String(128))
小开

这个线程和相关的 StackOverflow 线程解决方案使用 PostgreSQL 或其他特定于方言的类型。然而，通用支持可以很容易地在 SQLAlchemy 中实现，这也与 Alembic 迁移兼容。

如果后端不支持 Enum，SQLAlchemy 和 alembic 可以方便地对 varchar 和类似类型强制约束，以模拟枚举列类型。

首先，在声明自定义 SQLAlchemyEnum 列类型的任何位置导入 Python 枚举、 SQLAlchemyEnum 和 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,
)

当您运行 Alembic alembic 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,
...

要解决这个问题，导入 SQLAlchemyEnum 类，并在 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 Enum 类型，而不是使用 Alembic 尝试重新声明它。例如在 def upgrade() -> None:中

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

提交答案