SqlAlchemy：Postgresql 自定义类型的数组

Question

所以在我的 postgres 数据库中，我有以下自定义类型：

create type my_pg_type as (  
    sting_id varchar(32),
    time_diff interval,
    multiplier integer
);

更复杂的是，这被用作数组：

alter table my_table add column my_keys my_pg_type [];

我想用 SQLAlchemy (0.6.4) 映射这个！！

（为灵药道歉）

from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.types import Enum
from elixir import Entity, Field    

class MyTable(Entity):
    # -- snip --
    my_keys = Field(ARRAY(Enum))

我知道上面的“枚举”不正确。

对于从数据库返回的该数组列的值的示例，我已在ARRAY.result_processor(self, dialect, coltype) 中的值下方显示：

class ARRAY(sqltypes.MutableType, sqltypes.Concatenable, sqltypes.TypeEngine):
    # -- snip --  
    def result_processor(self, dialect, coltype):
        item_proc = self.item_type.result_processor(dialect, coltype)
        if item_proc:
            def convert_item(item):
                if isinstance(item, list):
                    return [convert_item(child) for child in item]
                else:
                    return item_proc(item)
        else:
            def convert_item(item):
                if isinstance(item, list):
                    return [convert_item(child) for child in item]
                else:
                    return item
        def process(value):
            if value is None:
                return value
            """
            # sample value:
             >>> value
            '{"(key_1,07:23:00,0)","(key_2,01:00:00,20)"}'
            """
            return [convert_item(item) for item in value]
        return process

所以上面的process 函数错误地拆分了字符串，假设它已经是一个列表。

到目前为止，我已经成功地将 ARRAY 子类化以正确拆分字符串，而不是 Enum，我尝试编写自己的类型（实现 Unicode）来重新创建 (string, timedelta, integer) 元组，但遇到了很多困难，特别是 interval 到 Python timedelta 的正确转换。

我在这里发帖以防我错过了一个明显的先例？

Answer 1

更新请参阅底部的配方以了解解决方法

我编写了一些示例代码来查看 psycopg2 在这里做什么，这完全在他们的范围内 - psycopg2 根本没有将值解释为数组。 psycopg2 需要能够在返回时解析出 ARRAY，因为 SQLA 的 ARRAY 类型假设至少已经完成了这么多工作。当然，您可以破解 SQLAlchemy 的 ARRAY，这意味着基本上不使用它，而是使用解析出 psycopg2 回馈给我们的特定字符串值的东西。

但是这里也发生了我们甚至没有了解 psycopg2 的转换时间增量的机制，这是 SQLAlchemy 通常不必担心的。在这种情况下，我觉得 DBAPI 的功能没有得到充分利用，而 psycopg2 是一个非常强大的 DBAPI。

所以我建议你在http://initd.org/psycopg/docs/extensions.html#database-types-casting-functions 使用 psycopg2 的自定义类型机制。

如果你想邮寄他们的mailing list，这里有一个测试用例：

import psycopg2

conn = psycopg2.connect(host="localhost", database="test", user="scott", password="tiger")
cursor = conn.cursor()
cursor.execute("""
create type my_pg_type as (  
    string_id varchar(32),
    time_diff interval,
    multiplier integer
)
""")

cursor.execute("""
    CREATE TABLE my_table (
        data my_pg_type[]
    )
""")

cursor.execute("insert into my_table (data) "
            "values (CAST(%(data)s AS my_pg_type[]))", 
            {'data':[("xyz", "'1 day 01:00:00'", 5), ("pqr", "'1 day 01:00:00'", 5)]})

cursor.execute("SELECT * from my_table")
row = cursor.fetchone()
assert isinstance(row[0], (tuple, list)), repr(row[0])

PG 的类型注册支持全局注册。您还可以在 SQLAlchemy 中使用 0.6 中的 pool listener 或 0.7 及更高版本中的 connect event 在每个连接的基础上注册类型。

更新 - 由于https://bitbucket.org/zzzeek/sqlalchemy/issue/3467/array-of-enums-does-not-allow-assigning，我可能会建议人们现在使用这种解决方法类型，直到 psycopg2 对此添加更多内置支持：

class ArrayOfEnum(ARRAY):

    def bind_expression(self, bindvalue):
        return sa.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(",")

        def process(value):
            return super_rp(handle_raw_string(value))
        return process

Answer 2

查看sqlalchemy_utils 文档：

CompositeType provides means to interact with
`PostgreSQL composite types`_. Currently this type features:

* Easy attribute access to composite type fields
* Supports SQLAlchemy TypeDecorator types
* Ability to include composite types as part of PostgreSQL arrays
* Type creation and dropping

用法：

from collections import OrderedDict

import sqlalchemy as sa
from sqlalchemy_utils import Composite, CurrencyType


class Account(Base):
    __tablename__ = 'account'
    id = sa.Column(sa.Integer, primary_key=True)
    balance = sa.Column(
        CompositeType(
            'money_type',
            [
                sa.Column('currency', CurrencyType),
                sa.Column('amount', sa.Integer)
            ]
        )
    )

复合数组：

from sqlalchemy_utils import CompositeArray


class Account(Base):
    __tablename__ = 'account'
    id = sa.Column(sa.Integer, primary_key=True)
    balances = sa.Column(
        CompositeArray(
            CompositeType(
                'money_type',
                [
                    sa.Column('currency', CurrencyType),
                    sa.Column('amount', sa.Integer)
                ]
            )
        )
    )