I solve this problem with new custom Query class:
class HieSession(Session):
def __init__(self, *args, **kwargs):
super(HieSession, self).__init__(*args, **kwargs)
self._query_cls = HieQuery
class HieQuery(Query):
def instances(self, cursor, __context=None):
"""Given a ResultProxy cursor as returned by connection.execute(),
return an ORM result as an iterator.
e.g.::
result = engine.execute("select * from users")
for u in session.query(User).instances(result):
print u
"""
session = self.session
context = __context
if context is None:
context = QueryContext(self)
context.runid = _new_runid()
filter_fns = [ent.filter_fn
for ent in self._entities]
filtered = id in filter_fns
single_entity = filtered and len(self._entities) == 1
if filtered:
if single_entity:
filter_fn = id
else:
def filter_fn(row):
return tuple(fn(x) for x, fn in zip(row, filter_fns))
custom_rows = single_entity and \
self._entities[0].mapper.dispatch.append_result
(process, labels) = \
zip(*[
query_entity.row_processor(self, context,
custom_rows)
for query_entity in self._entities
])
while True:
context.progress = {}
context.partials = {}
if self._yield_per:
fetch = cursor.fetchmany(self._yield_per)
if not fetch:
break
else:
fetch = cursor.fetchall()
if custom_rows:
rows = []
for row in fetch:
process[0](row, rows)
elif single_entity:
### ADDED BEGIN
#rows = [process[0](row, None) for row in fetch]
rows = []
for row in fetch:
instance = process[0](row, None)
row_tuple = tuple(row)
level = row_tuple[-3]
instance.level = level
rows.append(instance)
### ADDED END
else:
rows = [util.NamedTuple([proc(row, None) for proc in
process],
labels) for row in fetch]
if filtered:
rows = util.unique_list(rows, filter_fn)
if context.refresh_state and self._only_load_props \
and context.refresh_state in context.progress:
context.refresh_state.commit(
context.refresh_state.dict, self._only_load_props)
context.progress.pop(context.refresh_state)
session._finalize_loaded(context.progress)
for ii, (dict_, attrs) in context.partials.iteritems():
ii.commit(dict_, attrs)
for row in rows:
yield row
if not self._yield_per:
break
in HieQuery only added this code:
#rows = [process[0](row, None) for row in fetch]
rows = []
for row in fetch:
instance = process[0](row, None)
row_tuple = tuple(row)
level = row_tuple[-3]
instance.level = level
rows.append(instance)
HieSessionMaker = sessionmaker(class_=HieSession, bind=db.engine)
from app.lib.sqla_lib import HieSessionMaker
hie_session = HieSessionMaker()
and now I can wrote function for selecting with recursive from
sqla_hierarchy library and join User:
def get_hierarchy_objects(model_class, select_obj, starting_node=0,
limit=None, contains_eager_objs=None):
"""
returns hierarchy objects with joins and level from with recursive sql
model_class (Class) - class, e.g. Comment
select_obj (Select) - select object, e.g. select([Comment.id])
starting_node (int) - id for sub-trees
limit (int) - limit for result objects
contains_eager (tuple/list) - additional tables for join
example usage:
select all tree comments where content_id=1 with join user:
result = get_hierarchy_objects(Comment, select([Comment.__table__,
User.__table__], Comment.content_id==1,
from_obj=[outerjoin(Comment.__table__, User.__table__)]),
contains_eager_objs=[Comment.user])
"""
from app import hie_session
select_obj = select_obj.apply_labels()
h = hierarchy.Hierarchy(db.session, model_class.__table__, select_obj,
starting_node=starting_node)
h_compiled = h.compile()
hie_statement = h_compiled.__str__() % h_compiled.params
if limit:
hie_statement += ' LIMIT {0}'.format(limit)
result = hie_session.query(model_class).from_statement(hie_statement)
if contains_eager_objs:
result = result.options(contains_eager(*contains_eager_objs))
return result.all()
Thanks to all, SQLAlchemy very powerful! :)
On Thu, Oct 20, 2011 at 6:08 PM, Michael Bayer <[email protected]>wrote:
>
> On Oct 20, 2011, at 4:03 AM, Alex K wrote:
>
>
> result2 = db.session.query(non_primary_mapper).from_statement('SELECT
> test.id AS test_id, test.user_id AS test_user_id, test.reply_id AS
> test_reply_id, test.text AS test_text FROM test LEFT OUTER JOIN "user" ON
> "user".id = test.user_id LIMIT 1 OFFSET
> 0').options(contains_eager(Test.user)).all()
>
>
>
> There's absolutely no reason above you'd need to use a non primary mapper
> there. The primary mapper for Test is already Test.__table__. When you use
> from_statement(), the underlying table to which the class is mapped is
> entirely disregarded - the columns in the string statement are assigned to
> the Test object as well as the related User using a naming convention.
> You'd need to add all the "user" columns you'd like to load to the columns
> clause of the SELECT statement, using the convention
> <usertablename>_<columnname>. non primary mappers can't be used to add new
> columns to a mapping or to change the names of existing columns, so that
> isn't happening here either. The Test and User classes once mapped to
> their primary tables get their attribute names assigned and that's it.
>
> non primary mappers are almost entirely useless these days except for this
> one edge case involving relationship() to a non primary as the target, in
> order to emit custom SQL from that relationship, which is not the case here
> since you're writing the SQL by hand.
>
>
> --
> You received this message because you are subscribed to the Google Groups
> "sqlalchemy" group.
> To post to this group, send email to [email protected].
> To unsubscribe from this group, send email to
> [email protected].
> For more options, visit this group at
> http://groups.google.com/group/sqlalchemy?hl=en.
>
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected].
To unsubscribe from this group, send email to
[email protected].
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
