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 <mike...@zzzcomputing.com>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 sqlalchemy@googlegroups.com. > To unsubscribe from this group, send email to > sqlalchemy+unsubscr...@googlegroups.com. > 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 sqlalchemy@googlegroups.com. To unsubscribe from this group, send email to sqlalchemy+unsubscr...@googlegroups.com. For more options, visit this group at http://groups.google.com/group/sqlalchemy?hl=en.