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)
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)
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.
PostgreSQL supports sequences, and SQLAlchemy uses these as the default means of creating new primary key values for integer-based primary key columns.
The psycopg2 is over 2x faster than SQLAlchemy on small table. This behavior is expected as psycopg2 is a database driver for postgresql while SQLAlchemy is general ORM library.
In SQLAlchemy, the bindparam() construct has the ability to carry along the actual value that will be ultimately used at expression time.
Label is a class within the sqlalchemy. sql. elements module of the SQLAlchemy project.
Casting is very straightforward:
from sqlalchemy.dialects.postgresql import array, ARRAY
cast(array([]), ARRAY(Integer))
Without seeing what you've tried, I can only speculate that you tried to cast to array(Integer)
instead of ARRAY(Integer)
.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With