如何使用postgresql在sqlalchemy中显式转换数组文字的类型？

本文介绍了如何使用postgresql在sqlalchemy中显式转换数组文字的类型？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

尝试通过强制转换，type_coerce和type_强制转换文本数组类型后，没有想到我会问成功。

After attempting to cast a literal arrays type via cast, type_coerce, and type_ and not having any success thought I would ask.

from pprint import pprint from sqlalchemy import String, null, Integer, Column, ForeignKey, \ create_engine from sqlalchemy.dialects.postgresql import array from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import aliased, sessionmaker, relationship Base = declarative_base() temp_db_name = 'cf_LlAcKpxFHzOW' engine = create_engine('postgresql://localhost/{}'.format(temp_db_name)) class JobGroup(Base): __tablename__ = 'job_group' id = Column(Integer, primary_key=True) name = Column(String, nullable=False) parent_id = Column(Integer, ForeignKey("job_group.id")) parent = relationship("JobGroup", remote_side=[id]) def __init__(self, name, parent=None): self.name = name self.parent = parent def __repr__(self): return "JobGroup {} {}".format(self.id, self.name) Base.metadata.create_all(bind=engine) Session = sessionmaker() Session.configure(bind=engine) session = Session() gp_group = JobGroup(name="grandpa") p_group = JobGroup(name="parent", parent=gp_group) c_group = JobGroup(name="child", parent=p_group) session.add(gp_group) session.add(p_group) session.add(c_group) sessionmit() session.refresh(gp_group) session.refresh(p_group) session.refresh(c_group) # for jg in session.query(JobGroup).all(): # pprint(jg.__dict__) try: tree_parts = session.query( JobGroup.id, JobGroup.name, JobGroup.parent_id, array([]).label("ancestors") ).filter( JobGroup.parent_id == null() ).cte(name="tree_parts", recursive=True) jg_alias = aliased(JobGroup, name="jg") tree_parts_alias = aliased(tree_parts, name="tp") tree_parts = tree_parts.union_all( session.query( jg_alias.id, jg_alias.name, jg_alias.parent_id, (tree_parts_alias.c.ancestors + array([jg_alias.parent_id])).label("ancestors") ).filter(jg_alias.parent_id == tree_parts_alias.c.id) ) pprint(session.query(tree_parts).all()) finally: session.rollback() session.close_all() Base.metadata.drop_all(bind=engine)

这会导致postgres错误：

This results in the postgres error:

sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) cannot determine type of empty array LINE 2: ...p.name AS name, job_group.parent_id AS parent_id, ARRAY[] AS... ^ HINT: Explicitly cast to the desired type, for example ARRAY[]::integer[].

有很多方法可以解决此用例，例如预先填充祖先数组文字

There are ways to work around this for this use case, such as pre-populating the ancestors array literal with an integer that is an invalid parent_id, like -1.

推荐答案

铸造非常简单：

from sqlalchemy.dialects.postgresql import array, ARRAY cast(array([]), ARRAY(Integer))

没有看到您尝试过的内容，我只能推测您尝试转换为 array（Integer）而不是 ARRAY（Integer）。

Without seeing what you've tried, I can only speculate that you tried to cast to array(Integer) instead of ARRAY(Integer).