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 []; 

I 'd хотел бы сопоставить это с 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)) 

Я знаю '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

UPDATE Смотрите рецепт в нижней части для обходного

Я работал некоторый пример кода, чтобы увидеть, что psycopg2 делает здесь, и это хорошо в пределах их области - psycopg2 не интерпретировать значения как массив вообще. psycopg2 должен иметь возможность анализировать ARRAY, когда он возвращается, поскольку тип ARRAY SQLA предполагает, что, по крайней мере, это было сделано. Разумеется, вы можете взломать ARRAY SQLAlchemy, что в данном случае означает, что в основном не использует его вообще в пользу чего-то, что разбирает это значение строки, которое psycopg2 возвращает нам.

Но то, что здесь происходит, заключается в том, что мы даже не получаем механику psycopg2 для преобразования timedeltas, что-то SQLAlchemy, как правило, не нужно беспокоиться. В этом случае я чувствую, что возможности DBAPI недоиспользуются, а psycopg2 - очень способный DBAPI.

Поэтому я бы посоветовал вам работать с механикой типа psycopg2 по адресу http://initd.org/psycopg/docs/extensions.html#database-types-casting-functions.

Если вы хотите отправить их 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, используя pool listener в 0,6 или connect event в 0,7 и далее.

UPDATE - из-за 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

Checkout документацию 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) 
      ] 
     ) 
    ) 

массив 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) 
       ] 
      ) 
     ) 
    )