SqlAlchemy: array of Postgresql custom types

UPDATE See the recipe at the bottom for a workaround

I worked up some example code to see what psycopg2 is doing here, and this is well within their realm - psycopg2 is not interpreting the value as an array at all. psycopg2 needs to be able to parse out the ARRAY when it comes back as SQLA's ARRAY type assumes at least that much has been done. You can of course hack around SQLAlchemy's ARRAY, which here would mean basically not using it at all in favor of something that parses out this particular string value psycopg2 is giving us back.

But what's also happening here is that we aren't even getting at psycopg2's mechanics for converting timedeltas either, something SQLAlchemy normally doesn't have to worry about. In this case I feel like the facilities of the DBAPI are being under-utilized and psycopg2 is a very capable DBAPI.

So I'd advise you work with psycopg2's custom type mechanics over at http://initd.org/psycopg/docs/extensions.html#database-types-casting-functions.

If you want to mail their mailing list, here's a test case:

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's type registration supports global registration. You can also register the types on a per-connection basis within SQLAlchemy using the pool listener in 0.6 or connect event in 0.7 and further.

UPDATE - due to https://bitbucket.org/zzzeek/sqlalchemy/issue/3467/array-of-enums-does-not-allow-assigning I'm probably going to recommend people use this workaround type for now, until psycopg2 adds more built-in support for this:

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

Checkout the sqlalchemy_utils documentation:

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

Usage:

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

Array Of Composites:

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