from_statement and cte problem

32 views
Skip to first unread message

sector119

unread,
Jun 29, 2019, 11:23:57 AM6/29/19
to sqlalchemy
Hello,

I have Locality model with 'path' property to get path from 'root' of tree to current item, everything works ok, but
I can't get result as Locality instance list..
When I use 'object_session(self).query(Locality).from_statement(q).order_by(Locality.id)' I get 
sqlalchemy.exc.ArgumentError: from_statement accepts text(), select(), and union() objects only.

How can I adopt results to Locality model?


class Locality(Base):
    __tablename__ = 'localities'

    __table_args__ = {'schema': SYSTEM_SCHEMA}

    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey(SYSTEM_SCHEMA + '.localities.id'))
    name = Column(UnicodeText, nullable=False)
    type = Column(Integer, nullable=False)

    @property
    def path(self):
        def get_locality_path_q(locality_id):
            top_q = select([
                Locality.id,
                Locality.parent_id,
                Locality.name,
                Locality.type,
            ]).\
                where(Locality.id == locality_id).\
                cte(recursive=True)

            parents = aliased(top_q)

            locality_alias = aliased(Locality)

            q = top_q.union_all(
                select([
                    locality_alias.id,
                    locality_alias.parent_id,
                    locality_alias.name,
                    locality_alias.type
                ]).select_from(join(locality_alias, parents, locality_alias.id == parents.c.parent_id))
            )

            # return object_session(self).query(q).order_by(q.c.id)
            return object_session(self).query(Locality).from_statement(q).order_by(Locality.id)

        return get_locality_path_q(self.id)

Mike Bayer

unread,
Jun 29, 2019, 6:11:21 PM6/29/19
to noreply-spamdigest via sqlalchemy
above, "q" is a CTE, not a SELECT, meaning it's something you can select FROM, like a table.  Call select() to SELECT from it:

            return (
                object_session(self)
                .query(Locality)
                .from_statement(q.select().order_by(q.c.id))
            )


additionally, from_statement() does not allow further changes to the statement and the ORDER BY must be in terms of thing you are selecting from, in this case q.c.id

your code would be easier if you made use of top_q as a FROM object rather than a statement:

            lcte = aliased(Locality, q)

            return (
                object_session(self)
                .query(lcte)
                .order_by(lcte.id)
            )


There's not too many doc examples of how aliased() works with FROM clauses but the basic idea is at:



A little more on aliased() with CTE is written more for query.cte() which you could also be using here:







--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
 
 
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups "sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.
To post to this group, send email to sqlal...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

sector119

unread,
Jun 30, 2019, 5:36:57 AM6/30/19
to sqlalchemy
Nice, thanks a lot, Mike, now it works as expected

@property
def path(self):
    session = object_session(self)

    def get_locality_path_q(locality_id):
        parent_q = session.query(Locality).filter(Locality.id == locality_id).cte(recursive=True)

        parents = aliased(parent_q)

        locality_alias = aliased(Locality)

        q = parent_q.union_all(
            session.query(locality_alias).join(parents, locality_alias.id == parents.c.parent_id)
        )

        cte = aliased(Locality, q)

        return session.query(cte).order_by(cte.id)

    return get_locality_path_q(self.id)

воскресенье, 30 июня 2019 г., 1:11:21 UTC+3 пользователь Mike Bayer написал:
To unsubscribe from this group and stop receiving emails from it, send an email to sqlal...@googlegroups.com.

Mike Bayer

unread,
Jun 30, 2019, 1:26:42 PM6/30/19
to noreply-spamdigest via sqlalchemy


On Sun, Jun 30, 2019, at 5:37 AM, sector119 wrote:
Nice, thanks a lot, Mike, now it works as expected

that's great.  the docs are not good here, there's not enough discussion of "aliased()" , from_statement() and what they do, also I'm changing some details of how they do their work for 1.4 in any case so documentation efforts will be needed.



To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.
To post to this group, send email to sqlal...@googlegroups.com.

sector119

unread,
Jul 1, 2019, 4:20:56 PM7/1/19
to sqlalchemy
Mike, is it possible to load "@property" as subquery? Maybe as ARRAY[path_item, path_item, ...]

воскресенье, 30 июня 2019 г., 20:26:42 UTC+3 пользователь Mike Bayer написал:

Mike Bayer

unread,
Jul 1, 2019, 5:03:20 PM7/1/19
to noreply-spamdigest via sqlalchemy
I don't know what you mean.  is there SQL you know you want or is that the part you want to figure out?
To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.
To post to this group, send email to sqlal...@googlegroups.com.

sector119

unread,
Jul 1, 2019, 5:20:08 PM7/1/19
to sqlalchemy
For example I have a property 'path' with CTE like this and it wouldn't return set of rows, but scalar value as array[] of rows
would it be possible to load this property as subquery() ?

Of course I can include that CTE query in my query(Locality) using subquery(), but it would be cool if I'll be able to "undefer" that property somehow like relation )

@property
def path(self):
    session = object_session(self)

    def get_locality_path_q(locality_id):
        parent_q = session.query(Locality).filter(Locality.id == locality_id).cte(recursive=True)

        parents = aliased(parent_q)

        locality_alias = aliased(Locality)

        q = parent_q.union_all(
            session.query(locality_alias).join(parents, locality_alias.id == parents.c.parent_id)
        )

        cte = aliased(Locality, q)

        return session.query(cte).order_by(cte.id)

    return get_locality_path_q(self.id)

вторник, 2 июля 2019 г., 0:03:20 UTC+3 пользователь Mike Bayer написал:

Mike Bayer

unread,
Jul 1, 2019, 10:53:40 PM7/1/19
to noreply-spamdigest via sqlalchemy


On Mon, Jul 1, 2019, at 5:20 PM, sector119 wrote:
For example I have a property 'path' with CTE like this and it wouldn't return set of rows, but scalar value as array[] of rows
would it be possible to load this property as subquery() ?

I don't know from a SQL perspective you'd likely have to use some PG array function that does that, I don't really use those functions because I don't really buy into PG's SQL extension culture.  But anyway, if you can build a correlated scalar subquery that returns what you want, use column_property() and that will get it for you.

I can get the basic idea using func.array():

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import column_property
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

Base = declarative_base()


class A(Base):
    __tablename__ = "a"

    id = Column(Integer, primary_key=True)
    data = Column(String)
    bs = relationship("B")


class B(Base):
    __tablename__ = "b"
    id = Column(Integer, primary_key=True)
    a_id = Column(ForeignKey("a.id"))
    data = Column(String)


A.b_data = column_property(
    func.array(select([B.data]).where(B.a_id == A.id).as_scalar())
)

e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)

s = Session(e)

s.add_all(
    [A(bs=[B(data="b1"), B(data="b2")]), A(bs=[B(data="b3"), B(data="b4")])]
)
s.commit()

for a1 in s.query(A):
    print(a1.b_data)





To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+...@googlegroups.com.
To post to this group, send email to sqlal...@googlegroups.com.
Reply all
Reply to author
Forward
0 new messages