Here's a stripped down version in a single file. It shows only a single 
occasion of 'result' object appearing. The longer version above shows all.


On Sunday, July 19, 2015 at 9:09:01 PM UTC+2, Dirk Makowski wrote:
>
> Thanks for looking into it. Sorry about 7z, it's an archive. Please see 
> the revised attachment.
>
> In principle, I'm converting the result from a query into a list of dicts.
>
> data = []
> rs = sess.query(...)
> for r in rs:
>    data.append(dictate(r))
>
> In function dictate() is the call to inspect(). Until SA 0.9.9 it worked 
> fine. With version 1.0.6 sometimes the call to inspect() fails with 
> honestly that error message. It depends on what was queried:
> - a single ORM object: OK
> - hand-made SQL (sa.text()): OK
> - several joined ORM objects: FAIL
> - mixture of ORM objects and columns: FAIL
>
> Attached script goes through these situations and shows exactly what 
> happens. 
>
>
> On Sunday, July 19, 2015 at 6:28:07 PM UTC+2, Michael Bayer wrote:
>>
>>  
>>
>> On 7/19/15 11:58 AM, Dirk Makowski wrote:
>>  
>> Hello all, 
>>
>>  some time ago I wrote a function to mogrify SA query results into a 
>> list of dicts. It uses the inspector to determine columns etc. Up until SA 
>> 0.9.9 it worked well. However, after an upgrade of SA to 1.0.6 in some 
>> circumstances I get this error:
>>
>>   No inspection system is available for object of type <class 
>> 'sqlalchemy.util._collections.result'>
>>  
>>  What previously had been a Python type or ORM type now is an ominous 
>> 'result'. 
>>
>>  Hopefully you can give some pointers about what it is and how I could 
>> fix my function. A test case with more info in the sources is attached. 
>> Apart from SA it does have no external dependencies.
>>  
>>
>> The error is that you are calling inspect() on something that does not 
>> support inspection.  There's no object in util._collections called "result" 
>> so I don't know what that is.
>>
>>  
>
>> Also I don't know what a "7z" file is.  Can you please send a single, 
>> very succinct .py file with a simple illustration of your error?    Thanks.
>>
>>
>>  
>>  Thank you,
>>
>>  Dirk
>>
>>  -- 
>> 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.
>> Visit this group at http://groups.google.com/group/sqlalchemy.
>> For more options, visit https://groups.google.com/d/optout.
>>
>>
>>  

-- 
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+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
#!/usr/bin/env python

# ==================================================================
#  This is just setup. The real action is at the bottom
# ==================================================================

# from pprint import pprint
import sqlalchemy as sa
from sqlalchemy.exc import NoInspectionAvailable
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import sqlalchemy.util._collections


DbEngine = sa.create_engine('sqlite:///:memory:', echo=False)
DbBase = declarative_base()
DbSession = sessionmaker(bind=DbEngine)


class Smeagol(DbBase):
    __tablename__ = 'smeagol'

    id = sa.Column(sa.Integer(), nullable=False, primary_key=True)
    who = sa.Column(sa.Unicode(255), nullable=False)

    def __repr__(self):
        return "<{cls}(id={id}, who='{who}')>".format(
            cls=self.__class__.__name__, id=self.id, who=self.who)


class Deagol(DbBase):
    __tablename__ = 'deagol'

    id = sa.Column(sa.Integer(), nullable=False, primary_key=True)
    who_id = sa.Column(sa.Integer(), nullable=False)
    what = sa.Column(sa.Unicode(255), nullable=False)
    whom_id = sa.Column(sa.Integer(), nullable=True)

    def __repr__(self):
        return "<{cls}(id={id}, who_id={who}, what='{what}', whom_id={whom})>".format(
            cls=self.__class__.__name__, id=self.id, who=self.who_id,
            what=self.what, whom=self.whom_id)


def is_1xx():
    return sa.__version__.startswith('1')


def bootstrap(sess):
    DbBase.metadata.create_all(DbEngine)

    # Yeah, I could have used orm relationships to init, but I didn't
    me = Smeagol(who='Me')
    sess.add(me)
    myself = Smeagol(who='Myself')
    sess.add(myself)
    i = Smeagol(who='I')
    sess.add(i)
    sess.flush()

    sess.add(Deagol(who_id=i.id, what='hate', whom_id=myself.id))
    sess.add(Deagol(who_id=i.id, what="don't like", whom_id=me.id))
    sess.add(Deagol(who_id=myself.id, what="is bff of", whom_id=me.id))
    # Yes, I want a missing right side
    sess.add(Deagol(who_id=me.id, what="wtf? no SO?", whom_id=None))
    sess.flush()


def perform_query(sess):
    who = sa.orm.aliased(Smeagol, name='Who')
    what = sa.orm.aliased(Deagol, name='What')
    whom = sa.orm.aliased(Smeagol, name='Whom')

    rs = sess.query(
        who, what, whom
    ).outerjoin(
        what, what.who_id == who.id
    ).outerjoin(
        whom, what.whom_id == whom.id
    ).order_by(
        who.who.asc()
    )
    return rs


# ==================================================================
#  Real action begins here
# ==================================================================

def process_keyed_tuple(kt):
    for i in kt:
        if i is not None:
            sa.inspect(i)  # discard result, just show that inspect() works


def process_object(o):
    sa.inspect(o)  # discard result, just show that inspect() works


def dictate(r):
    if is_1xx():
        # Need to use name, since there is no class 'result'
        assert type(r).__name__ == 'result'
    else:
        assert type(r) == sqlalchemy.util._collections.KeyedTuple

    # KeyedTuples are recognized by the real function dictate(), iterated over
    # and each item gets inspected
    if isinstance(r, sqlalchemy.util._collections.KeyedTuple):
        process_keyed_tuple(r)
    else:
        # This type is unknown, then treated as an object and inspected(),
        # which fails.
        process_object(r)


def main():
    sess = DbSession()
    bootstrap(sess)

    rs = perform_query(sess)
    for r in rs:
        dictate(r)  # results not needed here

    print('Finished.')


if __name__ == '__main__':
    main()

Reply via email to