python - 带有 SQLAlchemy 的 Postgres 中的枚举数组

Question

在过去的一年里，我一直在 postgres 和 SQLAlchemy 中成功使用枚举数组，如下所示:

class MyModel(BaseModel):
    enum_field = Column(postgresql.ARRAY(EnumField(MyEnum, native_enum=False)))

EnumField 来自 sqlalchemy_enum34库，一个围绕内置枚举的小包装器，它使用 Python 枚举作为 Python 表示而不是字符串。

尽管 docs say ，不支持枚举数组，我猜它起作用了，因为我选择了“native_enum=False”。最近我注意到它不再工作了，我认为这是由于从 SQLA 1.0 升级到 1.1，但我不确定。

问题是，它生成了无效的 DQL:

CREATE TABLE my_model (
    enum_field VARCHAR(5)[3] NOT NULL CHECK (contexts IN ('ONE', 'TWO', 'THREE'))
)

我得到的错误是:

ERROR:  malformed array literal: "ONE"
DETAIL:  Array value must start with "{" or dimension information.

知道如何取回我的枚举数组吗？
顺便说一句:当它工作时，实际上没有创建 CHECK 约束，只是一个 varying 数组。我可以接受，只要我可以在我的 Python 代码中使用枚举(例如 query.filter(enum_field==MyEnum.ONE))

Answer 1

我在 SqlAlchemy 源代码中找到了很好的解决方法:

import re

from sqlalchemy import TypeDecorator, cast
from sqlalchemy.dialects.postgresql import ARRAY


class ArrayOfEnum(TypeDecorator):

    impl = ARRAY

    def bind_expression(self, bindvalue):
        return cast(bindvalue, self)

    def result_processor(self, dialect, coltype):
        super_rp = super(ArrayOfEnum, self).result_processor(dialect, coltype)

        def handle_raw_string(value):
            inner = re.match(r"^{(.*)}$", value).group(1)

            return inner.split(",") if inner else []

        def process(value):
            if value is None:
                return None

            return super_rp(handle_raw_string(value))

        return process

现在:

achievements = Column(ArrayOfEnum(Enum(AchievementsType)))

然后:

career.achievements = [AchievementsType.world, AchievementsType.local]