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

【问题标题】：How to explicitly cast type of array literal in sqlalchemy using postgresql?如何使用 postgresql 在 sqlalchemy 中显式转换数组文字的类型？
【发布时间】：2018-04-14 07:27:24
【问题描述】：

在尝试通过 cast、type_coerce 和 type_ 转换文字数组类型并且没有任何成功之后，我想我会问。

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)
session.commit()
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 错误：

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

对于这个用例，有一些方法可以解决这个问题，例如用一个无效的 parent_id 整数预填充祖先数组字面量，例如 -1。

【问题讨论】：

标签： python arrays postgresql sqlalchemy

【解决方案1】：

铸造非常简单：

from sqlalchemy.dialects.postgresql import array, ARRAY

cast(array([]), ARRAY(Integer))

在没有看到您尝试过的情况下，我只能推测您尝试转换为 array(Integer) 而不是 ARRAY(Integer)。

【讨论】：