arrays – SqlAlchemy：Postgresql自定义类型的数组
|
所以在我的postgres DB中我有以下自定义类型：
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))
我知道’Enum’在上面是不正确的. 有关从该数组的数据库列返回的值的示例,我在下面显示了ARRAY.result_processor(self,dialect,coltype)中的值： class ARRAY(sqltypes.MutableType,sqltypes.Concatenable,sqltypes.TypeEngine):
# -- snip --
def result_processor(self,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
所以上面的过程函数错误地分割了字符串,假设它已经是一个列表. 到目前为止,我已经成功地将ARRAY子类化为正确分割字符串,而不是Enum,我尝试编写自己的类型(实现Unicode)来重新创建(字符串,时间线,整数)元组,但是遇到了很多困难,特别是将间隔正确转换为Python timedelta. 我在这里张贴,以防我错过了一个明显的先例方法吗？ 解决方法
更新请参阅底部的配方以获取解决方法
我编写了一些示例代码来查看psycopg2在这里做了什么,这完全在他们的领域内 – psycopg2根本没有将值解释为数组.当它返回时,psycopg2需要能够解析出ARRAY,因为SQLA的ARRAY类型至少假定已经完成了很多.你当然可以破解SQLAlchemy的ARRAY,这里的意思是基本上不会使用它来支持解析psycopg2给我们回来的特定字符串值的东西. 但是这里也发生的事情是我们甚至没有获得psycopg2的转换timedeltas的机制,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),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",5)]})
cursor.execute("SELECT * from my_table")
row = cursor.fetchone()
assert isinstance(row[0],(tuple,list)),repr(row[0])
PG的类型注册支持全局注册.您还可以在SQLAlchemy中使用pool listener in 0.6或connect event in 0.7进行基于每个连接的类型注册. 更新 – 由于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,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
（编辑：李大同） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |