path: root/lib/sqlalchemy/orm/query.py

# orm/query.py
# Copyright (C) 2005, 2006, 2007, 2008 Michael Bayer mike_mp@zzzcomputing.com
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php

"""The Query class and support.

Defines the [sqlalchemy.orm.query#Query] class, the central construct used by
the ORM to construct database queries.

The ``Query`` class should not be confused with the
[sqlalchemy.sql.expression#Select] class, which defines database SELECT
operations at the SQL (non-ORM) level.  ``Query`` differs from ``Select`` in
that it returns ORM-mapped objects and interacts with an ORM session, whereas
the ``Select`` construct interacts directly with the database to return
iterable result sets.
"""

from itertools import chain
from sqlalchemy import sql, util, exceptions, logging
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql import expression, visitors, operators
from sqlalchemy.orm import mapper, object_mapper
from sqlalchemy.orm.mapper import _state_mapper
from sqlalchemy.orm import util as mapperutil
from sqlalchemy.orm import interfaces

__all__ = ['Query', 'QueryContext']


class Query(object):
    """Encapsulates the object-fetching operations provided by Mappers."""

    def __init__(self, class_or_mapper, session=None, entity_name=None):
        if isinstance(class_or_mapper, type):
            self.mapper = mapper.class_mapper(class_or_mapper, entity_name=entity_name)
        else:
            self.mapper = class_or_mapper.compile()
        self.select_mapper = self.mapper.get_select_mapper().compile()

        self._session = session

        self._with_options = []
        self._lockmode = None
        self._extension = self.mapper.extension
        self._entities = []
        self._order_by = False
        self._group_by = False
        self._distinct = False
        self._offset = None
        self._limit = None
        self._statement = None
        self._params = {}
        self._yield_per = None
        self._criterion = None
        self._joinable_tables = None
        self._having = None
        self._column_aggregate = None
        self._joinpoint = self.mapper
        self._aliases = None
        self._alias_ids = {}
        self._from_obj = self.table
        self._populate_existing = False
        self._version_check = False
        self._autoflush = True
        self._eager_loaders = util.Set(chain(*[mp._eager_loaders for mp in [m for m in self.mapper.iterate_to_root()]]))
        self._attributes = {}
        self._current_path = ()
        self._only_load_props = None
        self._refresh_instance = None
        
        self._adapter = self.select_mapper._clause_adapter
        
    def _no_criterion(self, meth):
        q = self._clone()

        if q._criterion or q._statement or q._from_obj is not self.table:
            util.warn(
                ("Query.%s() being called on a Query with existing criterion; "
                 "criterion is being ignored.") % meth)

        q._from_obj = self.table
        q._adapter = self.select_mapper._clause_adapter
        q._alias_ids = {}
        q._joinpoint = self.mapper
        q._statement = q._aliases = q._criterion = None
        q._order_by = q._group_by = q._distinct = False
        return q

    def _no_statement(self, meth):
        q = self._clone()
        if q._statement:
            raise exceptions.InvalidRequestError(
                ("Query.%s() being called on a Query with an existing full "
                 "statement - can't apply criterion.") % meth)
        return q

    def _clone(self):
        q = Query.__new__(Query)
        q.__dict__ = self.__dict__.copy()
        return q

    def _get_session(self):
        if self._session is None:
            return self.mapper.get_session()
        else:
            return self._session

    table = property(lambda s:s.select_mapper.mapped_table)
    primary_key_columns = property(lambda s:s.select_mapper.primary_key)
    session = property(_get_session)

    def _with_current_path(self, path):
        q = self._clone()
        q._current_path = path
        return q

    def yield_per(self, count):
        """Yield only ``count`` rows at a time.

        WARNING: use this method with caution; if the same instance is present
        in more than one batch of rows, end-user changes to attributes will be
        overwritten.

        In particular, it's usually impossible to use this setting with
        eagerly loaded collections (i.e. any lazy=False) since those
        collections will be cleared for a new load when encountered in a
        subsequent result batch.
        """

        q = self._clone()
        q._yield_per = count
        return q

    def get(self, ident, **kwargs):
        """Return an instance of the object based on the given identifier, or None if not found.

        The `ident` argument is a scalar or tuple of primary key column values
        in the order of the table def's primary key columns.
        """

        ret = self._extension.get(self, ident, **kwargs)
        if ret is not mapper.EXT_CONTINUE:
            return ret

        # convert composite types to individual args
        # TODO: account for the order of columns in the
        # ColumnProperty it corresponds to
        if hasattr(ident, '__composite_values__'):
            ident = ident.__composite_values__()

        key = self.mapper.identity_key_from_primary_key(ident)
        return self._get(key, ident, **kwargs)

    def load(self, ident, raiseerr=True, **kwargs):
        """Return an instance of the object based on the given identifier.

        If not found, raises an exception.  The method will **remove all
        pending changes** to the object already existing in the Session.  The
        `ident` argument is a scalar or tuple of primary key column values in
        the order of the table def's primary key columns.
        """

        ret = self._extension.load(self, ident, **kwargs)
        if ret is not mapper.EXT_CONTINUE:
            return ret
        key = self.mapper.identity_key_from_primary_key(ident)
        instance = self.populate_existing()._get(key, ident, **kwargs)
        if instance is None and raiseerr:
            raise exceptions.InvalidRequestError("No instance found for identity %s" % repr(ident))
        return instance

    def query_from_parent(cls, instance, property, **kwargs):
        """Return a new Query with criterion corresponding to a parent instance.

        Return a newly constructed Query object, with criterion corresponding
        to a relationship to the given parent instance.

        instance
          a persistent or detached instance which is related to class
          represented by this query.

         property
           string name of the property which relates this query's class to the
           instance.

         \**kwargs
           all extra keyword arguments are propigated to the constructor of
           Query.
        """

        mapper = object_mapper(instance)
        prop = mapper.get_property(property, resolve_synonyms=True)
        target = prop.mapper
        criterion = prop.compare(operators.eq, instance, value_is_parent=True)
        return Query(target, **kwargs).filter(criterion)
    query_from_parent = classmethod(query_from_parent)

    def autoflush(self, setting):
        q = self._clone()
        q._autoflush = setting
        return q

    def populate_existing(self):
        """Return a Query that will refresh all instances loaded.

        This includes all entities accessed from the database, including
        secondary entities, eagerly-loaded collection items.

        All changes present on entities which are already present in the
        session will be reset and the entities will all be marked "clean".

        This is essentially the en-masse version of load().
        """

        q = self._clone()
        q._populate_existing = True
        return q

    def with_parent(self, instance, property=None):
        """add a join criterion corresponding to a relationship to the given parent instance.

            instance
                a persistent or detached instance which is related to class represented
                by this query.

            property
                string name of the property which relates this query's class to the
                instance.  if None, the method will attempt to find a suitable property.

        currently, this method only works with immediate parent relationships, but in the
        future may be enhanced to work across a chain of parent mappers.
        """

        from sqlalchemy.orm import properties
        mapper = object_mapper(instance)
        if property is None:
            for prop in mapper.iterate_properties:
                if isinstance(prop, properties.PropertyLoader) and prop.mapper is self.mapper:
                    break
            else:
                raise exceptions.InvalidRequestError("Could not locate a property which relates instances of class '%s' to instances of class '%s'" % (self.mapper.class_.__name__, instance.__class__.__name__))
        else:
            prop = mapper.get_property(property, resolve_synonyms=True)
        return self.filter(prop.compare(operators.eq, instance, value_is_parent=True))

    def add_entity(self, entity, alias=None, id=None):
        """add a mapped entity to the list of result columns to be returned.

        This will have the effect of all result-returning methods returning a tuple
        of results, the first element being an instance of the primary class for this
        Query, and subsequent elements matching columns or entities which were
        specified via add_column or add_entity.

        When adding entities to the result, its generally desireable to add
        limiting criterion to the query which can associate the primary entity
        of this Query along with the additional entities.  The Query selects
        from all tables with no joining criterion by default.

            entity
                a class or mapper which will be added to the results.

            alias
                a sqlalchemy.sql.Alias object which will be used to select rows.  this
                will match the usage of the given Alias in filter(), order_by(), etc. expressions

            id
                a string ID matching that given to query.join() or query.outerjoin(); rows will be
                selected from the aliased join created via those methods.
        """
        q = self._clone()

        if isinstance(entity, type):
            entity = mapper.class_mapper(entity)
        if alias is not None:
            alias = mapperutil.AliasedClauses(entity.mapped_table, alias=alias)

        q._entities = q._entities + [(entity, alias, id)]
        return q

    def add_column(self, column, id=None):
        """Add a SQL ColumnElement to the list of result columns to be returned.

        This will have the effect of all result-returning methods returning a
        tuple of results, the first element being an instance of the primary
        class for this Query, and subsequent elements matching columns or
        entities which were specified via add_column or add_entity.

        When adding columns to the result, its generally desireable to add
        limiting criterion to the query which can associate the primary entity
        of this Query along with the additional columns, if the column is
        based on a table or selectable that is not the primary mapped
        selectable.  The Query selects from all tables with no joining
        criterion by default.

        column
          a string column name or sql.ColumnElement to be added to the results.
        """

        q = self._clone()

        # duck type to get a ClauseElement
        if hasattr(column, 'clause_element'):
            column = column.clause_element()

        # alias non-labeled column elements.
        if isinstance(column, sql.ColumnElement) and not hasattr(column, '_label'):
            column = column.label(None)

        q._entities = q._entities + [(column, None, id)]
        return q

    def options(self, *args):
        """Return a new Query object, applying the given list of
        MapperOptions.
        """

        return self._options(False, *args)

    def _conditional_options(self, *args):
        return self._options(True, *args)

    def _options(self, conditional, *args):
        q = self._clone()
        # most MapperOptions write to the '_attributes' dictionary,
        # so copy that as well
        q._attributes = q._attributes.copy()
        opts = [o for o in util.flatten_iterator(args)]
        q._with_options = q._with_options + opts
        if conditional:
            for opt in opts:
                opt.process_query_conditionally(q)
        else:
            for opt in opts:
                opt.process_query(q)
        return q

    def with_lockmode(self, mode):
        """Return a new Query object with the specified locking mode."""
        q = self._clone()
        q._lockmode = mode
        return q

    def params(self, *args, **kwargs):
        """add values for bind parameters which may have been specified in filter().

        parameters may be specified using \**kwargs, or optionally a single dictionary
        as the first positional argument.  The reason for both is that \**kwargs is
        convenient, however some parameter dictionaries contain unicode keys in which case
        \**kwargs cannot be used.
        """

        q = self._clone()
        if len(args) == 1:
            d = args[0]
            kwargs.update(d)
        elif len(args) > 0:
            raise exceptions.ArgumentError("params() takes zero or one positional argument, which is a dictionary.")
        q._params = q._params.copy()
        q._params.update(kwargs)
        return q
    
    def filter(self, criterion):
        """apply the given filtering criterion to the query and return the newly resulting ``Query``

        the criterion is any sql.ClauseElement applicable to the WHERE clause of a select.
        """

        if isinstance(criterion, basestring):
            criterion = sql.text(criterion)

        if criterion is not None and not isinstance(criterion, sql.ClauseElement):
            raise exceptions.ArgumentError("filter() argument must be of type sqlalchemy.sql.ClauseElement or string")

        if self._adapter is not None:
            criterion = self._adapter.traverse(criterion)
            
        q = self._no_statement("filter")
        if q._criterion is not None:
            q._criterion = q._criterion & criterion
        else:
            q._criterion = criterion
        return q

    def filter_by(self, **kwargs):
        """apply the given filtering criterion to the query and return the newly resulting ``Query``."""

        clauses = [self._joinpoint.get_property(key, resolve_synonyms=True).compare(operators.eq, value)
            for key, value in kwargs.iteritems()]

        return self.filter(sql.and_(*clauses))

    def _get_joinable_tables(self):
        if not self._joinable_tables or self._joinable_tables[0] is not self._from_obj:
            currenttables = [self._from_obj]
            def visit_join(join):
                currenttables.append(join.left)
                currenttables.append(join.right)
            visitors.traverse(self._from_obj, visit_join=visit_join, traverse_options={'column_collections':False, 'aliased_selectables':False})
            self._joinable_tables = (self._from_obj, currenttables)
            return currenttables
        else:
            return self._joinable_tables[1]

    def _join_to(self, keys, outerjoin=False, start=None, create_aliases=True):
        if start is None:
            start = self._joinpoint

        clause = self._from_obj

        currenttables = self._get_joinable_tables()

        # determine if generated joins need to be aliased on the left
        # hand side.  
        if self._adapter and not self._aliases:  # at the beginning of a join, look at leftmost adapter
            adapt_against = self._adapter.selectable
        elif start.select_table is not start.mapped_table: # in the middle of a join, look for a polymorphic mapper
            adapt_against = start.select_table
        else:
            adapt_against = None

        mapper = start
        alias = self._aliases
        for key in util.to_list(keys):
            if isinstance(key, interfaces.PropComparator):
                prop = key.property
            else:
                prop = mapper.get_property(key, resolve_synonyms=True)
                
            if prop._is_self_referential() and not create_aliases:
                raise exceptions.InvalidRequestError("Self-referential query on '%s' property requires create_aliases=True argument." % str(prop))

            if prop.select_table not in currenttables or create_aliases:
                if prop.secondary:
                    if create_aliases:
                        alias = mapperutil.PropertyAliasedClauses(prop,
                            prop.primary_join_against(mapper, adapt_against),
                            prop.secondary_join_against(mapper),
                            alias
                        )
                        crit = alias.primaryjoin
                        clause = clause.join(alias.secondary, crit, isouter=outerjoin).join(alias.alias, alias.secondaryjoin, isouter=outerjoin)
                    else:
                        crit = prop.primary_join_against(mapper, adapt_against)
                        clause = clause.join(prop.secondary, crit, isouter=outerjoin)
                        clause = clause.join(prop.select_table, prop.secondary_join_against(mapper), isouter=outerjoin)
                else:
                    if create_aliases:
                        alias = mapperutil.PropertyAliasedClauses(prop,
                            prop.primary_join_against(mapper, adapt_against), 
                            None,
                            alias
                        )
                        crit = alias.primaryjoin
                        clause = clause.join(alias.alias, crit, isouter=outerjoin)
                    else:
                        crit = prop.primary_join_against(mapper, adapt_against)
                        clause = clause.join(prop.select_table, crit, isouter=outerjoin)
            elif not create_aliases and prop.secondary is not None and prop.secondary not in currenttables:
                # TODO: this check is not strong enough for different paths to the same endpoint which
                # does not use secondary tables
                raise exceptions.InvalidRequestError("Can't join to property '%s'; a path to this table along a different secondary table already exists.  Use the `alias=True` argument to `join()`." % prop.key)

            mapper = prop.mapper

            if mapper.select_table is not mapper.mapped_table:
                adapt_against = mapper.select_table

        if create_aliases:
            return (clause, mapper, alias)
        else:
            return (clause, mapper, None)

    def _generative_col_aggregate(self, col, func):
        """apply the given aggregate function to the query and return the newly
        resulting ``Query``.
        """
        if self._column_aggregate is not None:
            raise exceptions.InvalidRequestError("Query already contains an aggregate column or function")
        q = self._no_statement("aggregate")
        q._column_aggregate = (col, func)
        return q

    def apply_min(self, col):
        """apply the SQL ``min()`` function against the given column to the
        query and return the newly resulting ``Query``.
        """
        return self._generative_col_aggregate(col, sql.func.min)

    def apply_max(self, col):
        """apply the SQL ``max()`` function against the given column to the
        query and return the newly resulting ``Query``.
        """
        return self._generative_col_aggregate(col, sql.func.max)

    def apply_sum(self, col):
        """apply the SQL ``sum()`` function against the given column to the
        query and return the newly resulting ``Query``.
        """
        return self._generative_col_aggregate(col, sql.func.sum)

    def apply_avg(self, col):
        """apply the SQL ``avg()`` function against the given column to the
        query and return the newly resulting ``Query``.
        """
        return self._generative_col_aggregate(col, sql.func.avg)

    def _col_aggregate(self, col, func):
        """Execute ``func()`` function against the given column.

        For performance, only use subselect if `order_by` attribute is set.
        """

        ops = {'distinct':self._distinct, 'order_by':self._order_by or None, 'from_obj':self._from_obj}

        if self._autoflush and not self._populate_existing:
            self.session._autoflush()

        if self._order_by is not False:
            s1 = sql.select([col], self._criterion, **ops).alias('u')
            return self.session.execute(sql.select([func(s1.corresponding_column(col))]), mapper=self.mapper).scalar()
        else:
            return self.session.execute(sql.select([func(col)], self._criterion, **ops), mapper=self.mapper).scalar()

    def min(self, col):
        """Execute the SQL ``min()`` function against the given column."""

        return self._col_aggregate(col, sql.func.min)

    def max(self, col):
        """Execute the SQL ``max()`` function against the given column."""

        return self._col_aggregate(col, sql.func.max)

    def sum(self, col):
        """Execute the SQL ``sum()`` function against the given column."""

        return self._col_aggregate(col, sql.func.sum)

    def avg(self, col):
        """Execute the SQL ``avg()`` function against the given column."""

        return self._col_aggregate(col, sql.func.avg)

    def order_by(self, criterion):
        """apply one or more ORDER BY criterion to the query and return the newly resulting ``Query``"""

        q = self._no_statement("order_by")

        if self._adapter:
            criterion = [expression._literal_as_text(o) for o in util.to_list(criterion) or []]
            criterion = self._adapter.copy_and_process(criterion)

        if q._order_by is False:
            q._order_by = util.to_list(criterion)
        else:
            q._order_by = q._order_by + util.to_list(criterion)
        return q

    def group_by(self, criterion):
        """apply one or more GROUP BY criterion to the query and return the newly resulting ``Query``"""

        q = self._no_statement("group_by")
        if q._group_by is False:
            q._group_by = util.to_list(criterion)
        else:
            q._group_by = q._group_by + util.to_list(criterion)
        return q

    def having(self, criterion):
        """apply a HAVING criterion to the quer and return the newly resulting ``Query``."""

        if isinstance(criterion, basestring):
            criterion = sql.text(criterion)

        if criterion is not None and not isinstance(criterion, sql.ClauseElement):
            raise exceptions.ArgumentError("having() argument must be of type sqlalchemy.sql.ClauseElement or string")

        if self._adapter is not None:
            criterion = self._adapter.traverse(criterion)

        q = self._no_statement("having")
        if q._having is not None:
            q._having = q._having & criterion
        else:
            q._having = criterion
        return q

    def join(self, prop, id=None, aliased=False, from_joinpoint=False):
        """Create a join against this ``Query`` object's criterion
        and apply generatively, retunring the newly resulting ``Query``.

        'prop' may be one of:
          * a string property name, i.e. "rooms"
          * a class-mapped attribute, i.e. Houses.rooms
          * a list containing a combination of any of the above.
          
        e.g.::
        
            session.query(Company).join('employees')
            session.query(Company).join(['employees', 'tasks'])
            session.query(Houses).join([Colonials.rooms, Room.closets])
        
        """

        return self._join(prop, id=id, outerjoin=False, aliased=aliased, from_joinpoint=from_joinpoint)

    def outerjoin(self, prop, id=None, aliased=False, from_joinpoint=False):
        """Create a left outer join against this ``Query`` object's criterion
        and apply generatively, retunring the newly resulting ``Query``.

        'prop' may be one of:
          * a string property name, i.e. "rooms"
          * a class-mapped attribute, i.e. Houses.rooms
          * a list containing a combination of any of the above.
          
        e.g.::
        
            session.query(Company).outerjoin('employees')
            session.query(Company).outerjoin(['employees', 'tasks'])
            session.query(Houses).outerjoin([Colonials.rooms, Room.closets])
        
        """

        return self._join(prop, id=id, outerjoin=True, aliased=aliased, from_joinpoint=from_joinpoint)

    def _join(self, prop, id, outerjoin, aliased, from_joinpoint):
        (clause, mapper, aliases) = self._join_to(prop, outerjoin=outerjoin, start=from_joinpoint and self._joinpoint or self.mapper, create_aliases=aliased)
        q = self._no_statement("join")
        q._from_obj = clause
        q._joinpoint = mapper
        q._aliases = aliases
        
        if aliases:
            q._adapter = sql_util.ClauseAdapter(aliases.alias).copy_and_chain(q._adapter)
        else:
            select_mapper = mapper.get_select_mapper()
            if select_mapper._clause_adapter:
                q._adapter = select_mapper._clause_adapter.copy_and_chain(q._adapter)

        a = aliases
        while a is not None:
            q._alias_ids.setdefault(a.mapper, []).append(a)
            q._alias_ids.setdefault(a.table, []).append(a)
            q._alias_ids.setdefault(a.alias, []).append(a)
            a = a.parentclauses

        if id:
            q._alias_ids[id] = aliases
        return q

    def reset_joinpoint(self):
        """return a new Query reset the 'joinpoint' of this Query reset
        back to the starting mapper.  Subsequent generative calls will
        be constructed from the new joinpoint.

        Note that each call to join() or outerjoin() also starts from
        the root.
        """

        q = self._no_statement("reset_joinpoint")
        q._joinpoint = q.mapper
        q._aliases = None
        if q.table not in q._get_joinable_tables():
            q._adapter = sql_util.ClauseAdapter(q._from_obj, equivalents=q.mapper._equivalent_columns)
        return q


    def select_from(self, from_obj):
        """Set the `from_obj` parameter of the query and return the newly
        resulting ``Query``.  This replaces the table which this Query selects
        from with the given table.


        `from_obj` is a single table or selectable.
        """

        new = self._no_criterion('select_from')
        if isinstance(from_obj, (tuple, list)):
            util.warn_deprecated("select_from() now accepts a single Selectable as its argument, which replaces any existing FROM criterion.")
            from_obj = from_obj[-1]

        if isinstance(from_obj, expression._SelectBaseMixin):
            # alias SELECTs and unions
            from_obj = from_obj.alias()

        new._from_obj = from_obj

        if new.table not in new._get_joinable_tables():
            new._adapter = sql_util.ClauseAdapter(new._from_obj, equivalents=new.mapper._equivalent_columns)
        return new

    def __getitem__(self, item):
        if isinstance(item, slice):
            start = item.start
            stop = item.stop
            # if we slice from the end we need to execute the query
            if (isinstance(start, int) and start < 0) or \
               (isinstance(stop, int) and stop < 0):
                return list(self)[item]
            else:
                res = self._clone()
                if start is not None and stop is not None:
                    res._offset = (self._offset or 0) + start
                    res._limit = stop - start
                elif start is None and stop is not None:
                    res._limit = stop
                elif start is not None and stop is None:
                    res._offset = (self._offset or 0) + start
                if item.step is not None:
                    return list(res)[None:None:item.step]
                else:
                    return res
        else:
            return list(self[item:item+1])[0]

    def limit(self, limit):
        """Apply a ``LIMIT`` to the query and return the newly resulting
        ``Query``.
        """

        return self[:limit]

    def offset(self, offset):
        """Apply an ``OFFSET`` to the query and return the newly resulting
        ``Query``.
        """

        return self[offset:]

    def distinct(self):
        """Apply a ``DISTINCT`` to the query and return the newly resulting
        ``Query``.
        """

        new = self._no_statement("distinct")
        new._distinct = True
        return new

    def all(self):
        """Return the results represented by this ``Query`` as a list.

        This results in an execution of the underlying query.
        """
        return list(self)


    def from_statement(self, statement):
        """Execute the given SELECT statement and return results.

        This method bypasses all internal statement compilation, and the
        statement is executed without modification.

        The statement argument is either a string, a ``select()`` construct,
        or a ``text()`` construct, and should return the set of columns
        appropriate to the entity class represented by this ``Query``.

        Also see the ``instances()`` method.

        """

        if isinstance(statement, basestring):
            statement = sql.text(statement)
        q = self._no_criterion('from_statement')
        q._statement = statement
        return q

    def first(self):
        """Return the first result of this ``Query`` or None if the result doesn't contain any row.

        This results in an execution of the underlying query.
        """

        if self._column_aggregate is not None:
            return self._col_aggregate(*self._column_aggregate)

        ret = list(self[0:1])
        if len(ret) > 0:
            return ret[0]
        else:
            return None

    def one(self):
        """Return the first result of this ``Query``, raising an exception if more than one row exists.

        This results in an execution of the underlying query.
        """

        if self._column_aggregate is not None:
            return self._col_aggregate(*self._column_aggregate)

        ret = list(self[0:2])

        if len(ret) == 1:
            return ret[0]
        elif len(ret) == 0:
            raise exceptions.InvalidRequestError('No rows returned for one()')
        else:
            raise exceptions.InvalidRequestError('Multiple rows returned for one()')

    def __iter__(self):
        context = self._compile_context()
        context.statement.use_labels = True
        if self._autoflush and not self._populate_existing:
            self.session._autoflush()
        return self._execute_and_instances(context)

    def _execute_and_instances(self, querycontext):
        result = self.session.execute(querycontext.statement, params=self._params, mapper=self.mapper, instance=self._refresh_instance)
        return self.iterate_instances(result, querycontext=querycontext)

    def instances(self, cursor, *mappers_or_columns, **kwargs):
        return list(self.iterate_instances(cursor, *mappers_or_columns, **kwargs))

    def iterate_instances(self, cursor, *mappers_or_columns, **kwargs):
        session = self.session

        context = kwargs.pop('querycontext', None)
        if context is None:
            context = QueryContext(self)

        context.runid = _new_runid()

        mappers_or_columns = tuple(self._entities) + mappers_or_columns
        tuples = bool(mappers_or_columns)

        if context.row_adapter:
            def main(context, row):
                return self.select_mapper._instance(context, context.row_adapter(row), None,
                    extension=context.extension, only_load_props=context.only_load_props, refresh_instance=context.refresh_instance
                )
        else:
            def main(context, row):
                return self.select_mapper._instance(context, row, None,
                    extension=context.extension, only_load_props=context.only_load_props, refresh_instance=context.refresh_instance
                )

        if tuples:
            process = []
            process.append(main)
            for tup in mappers_or_columns:
                if isinstance(tup, tuple):
                    (m, alias, alias_id) = tup
                    clauses = self._get_entity_clauses(tup)
                else:
                    clauses = alias = alias_id = None
                    m = tup

                if isinstance(m, type):
                    m = mapper.class_mapper(m)

                if isinstance(m, mapper.Mapper):
                    def x(m):
                        row_adapter = clauses is not None and clauses.row_decorator or (lambda row: row)
                        def proc(context, row):
                            return m._instance(context, row_adapter(row), None)
                        process.append(proc)
                    x(m)
                elif isinstance(m, (sql.ColumnElement, basestring)):
                    def y(m):
                        row_adapter = clauses is not None and clauses.row_decorator or (lambda row: row)
                        def proc(context, row):
                            return row_adapter(row)[m]
                        process.append(proc)
                    y(m)
                else:
                    raise exceptions.InvalidRequestError("Invalid column expression '%r'" % m)

        while True:
            context.progress = util.Set()

            if self._yield_per:
                fetch = cursor.fetchmany(self._yield_per)
                if not fetch:
                    return
            else:
                fetch = cursor.fetchall()

            if tuples:
                rows = util.OrderedSet()
                for row in fetch:
                    rows.add(tuple([proc(context, row) for proc in process]))
            else:
                rows = util.UniqueAppender([])
                for row in fetch:
                    rows.append(main(context, row))

            if context.refresh_instance and context.only_load_props and context.refresh_instance in context.progress:
                context.refresh_instance.commit(context.only_load_props)
                context.progress.remove(context.refresh_instance)

            for ii in context.progress:
                context.attributes.get(('populating_mapper', ii), _state_mapper(ii))._post_instance(context, ii)
                ii.commit_all()

            for row in rows:
                yield row

            if not self._yield_per:
                break

    def _get(self, key=None, ident=None, refresh_instance=None, lockmode=None, only_load_props=None):
        lockmode = lockmode or self._lockmode
        if not self._populate_existing and not refresh_instance and not self.mapper.always_refresh and lockmode is None:
            try:
                return self.session.identity_map[key]
            except KeyError:
                pass

        if ident is None:
            if key is not None:
                ident = key[1]
        else:
            ident = util.to_list(ident)

        q = self
        if ident is not None:
            q = q._no_criterion('get')
            params = {}
            (_get_clause, _get_params) = self.select_mapper._get_clause
            q = q.filter(_get_clause)
            for i, primary_key in enumerate(self.primary_key_columns):
                try:
                    params[_get_params[primary_key].key] = ident[i]
                except IndexError:
                    raise exceptions.InvalidRequestError("Could not find enough values to formulate primary key for query.get(); primary key columns are %s" % ', '.join(["'%s'" % str(c) for c in self.primary_key_columns]))
            q = q.params(params)

        if lockmode is not None:
            q = q.with_lockmode(lockmode)
        q = q._select_context_options(populate_existing=bool(refresh_instance), version_check=(lockmode is not None), only_load_props=only_load_props, refresh_instance=refresh_instance)
        q._order_by = None
        try:
            # call using all() to avoid LIMIT compilation complexity
            return q.all()[0]
        except IndexError:
            return None

    def _nestable(self, **kwargs):
        """Return true if the given statement options imply it should be nested."""

        return (kwargs.get('limit') is not None or kwargs.get('offset') is not None or kwargs.get('distinct', False))

    def count(self, whereclause=None, params=None, **kwargs):
        """Apply this query's criterion to a SELECT COUNT statement.

        the whereclause, params and \**kwargs arguments are deprecated.  use filter()
        and other generative methods to establish modifiers.
        """

        q = self
        if whereclause is not None:
            q = q.filter(whereclause)
        if params is not None:
            q = q.params(params)
        q = q._legacy_select_kwargs(**kwargs)
        return q._count()

    def _count(self):
        """Apply this query's criterion to a SELECT COUNT statement.

        this is the purely generative version which will become
        the public method in version 0.5.
        """

        whereclause = self._criterion

        context = QueryContext(self)
        from_obj = self._from_obj

        if self._nestable(**self._select_args()):
            s = sql.select([self.table], whereclause, from_obj=from_obj, **self._select_args()).alias('getcount').count()
        else:
            primary_key = self.primary_key_columns
            s = sql.select([sql.func.count(list(primary_key)[0])], whereclause, from_obj=from_obj, **self._select_args())
        if self._autoflush and not self._populate_existing:
            self.session._autoflush()
        return self.session.scalar(s, params=self._params, mapper=self.mapper)

    def compile(self):
        """compiles and returns a SQL statement based on the criterion and conditions within this Query."""
        return self._compile_context().statement

    def _compile_context(self):

        context = QueryContext(self)

        if self._statement:
            self._statement.use_labels = True
            context.statement = self._statement
            return context

        whereclause = self._criterion

        from_obj = self._from_obj

        # if the query's ClauseAdapter is present, and its
        # specifically adapting against a modified "select_from"
        # argument, apply adaptation to the
        # individually selected columns as well as "eager" clauses added;
        # otherwise its currently not needed
        if self._adapter and self.table not in self._get_joinable_tables():
            adapter = self._adapter
        else:
            adapter = None

        adapter = self._adapter
        
        # TODO: mappers added via add_entity(), adapt their queries also,
        # if those mappers are polymorphic

        order_by = self._order_by
        if order_by is False:
            order_by = self.select_mapper.order_by
        if order_by is False:
            order_by = []
            if self.table.default_order_by() is not None:
                order_by = self.table.default_order_by()
            if from_obj.default_order_by() is not None:
                order_by = from_obj.default_order_by()

        try:
            for_update = {'read':'read','update':True,'update_nowait':'nowait',None:False}[self._lockmode]
        except KeyError:
            raise exceptions.ArgumentError("Unknown lockmode '%s'" % self._lockmode)

        # if single-table inheritance mapper, add "typecol IN (polymorphic)" criterion so
        # that we only load the appropriate types
        if self.select_mapper.single and self.select_mapper.polymorphic_on is not None and self.select_mapper.polymorphic_identity is not None:
            whereclause = sql.and_(whereclause, self.select_mapper.polymorphic_on.in_([m.polymorphic_identity for m in self.select_mapper.polymorphic_iterator()]))

        context.from_clause = from_obj

        # give all the attached properties a chance to modify the query
        # TODO: doing this off the select_mapper.  if its the polymorphic mapper, then
        # it has no relations() on it.  should we compile those too into the query ?  (i.e. eagerloads)
        for value in self.select_mapper.iterate_properties:
            if self._only_load_props and value.key not in self._only_load_props:
                continue
            context.exec_with_path(self.select_mapper, value.key, value.setup, context, only_load_props=self._only_load_props)

        # additional entities/columns, add those to selection criterion
        for tup in self._entities:
            (m, alias, alias_id) = tup
            clauses = self._get_entity_clauses(tup)
            if isinstance(m, mapper.Mapper):
                for value in m.iterate_properties:
                    context.exec_with_path(m, value.key, value.setup, context, parentclauses=clauses)
            elif isinstance(m, sql.ColumnElement):
                if clauses is not None:
                    m = clauses.aliased_column(m)
                context.secondary_columns.append(m)

        if self._eager_loaders and self._nestable(**self._select_args()):
            # eager loaders are present, and the SELECT has limiting criterion
            # produce a "wrapped" selectable.

            # ensure all 'order by' elements are ClauseElement instances
            # (since they will potentially be aliased)
            # locate all embedded Column clauses so they can be added to the
            # "inner" select statement where they'll be available to the enclosing
            # statement's "order by"

            cf = util.Set()
            if order_by:
                order_by = [expression._literal_as_text(o) for o in util.to_list(order_by) or []]
                for o in order_by:
                    cf.update(sql_util.find_columns(o))

            if adapter:
                # TODO: make usage of the ClauseAdapter here to create the list
                # of primary columns
                context.primary_columns = [from_obj.corresponding_column(c) or c for c in context.primary_columns]
                cf = [from_obj.corresponding_column(c) or c for c in cf]

            s2 = sql.select(context.primary_columns + list(cf), whereclause, from_obj=context.from_clause, use_labels=True, correlate=False, order_by=util.to_list(order_by), **self._select_args())

            s3 = s2.alias()

            context.row_adapter = mapperutil.create_row_adapter(s3, self.table)

            statement = sql.select([s3] + context.secondary_columns, for_update=for_update, use_labels=True)

            if context.eager_joins:
                eager_joins = sql_util.ClauseAdapter(s3).traverse(context.eager_joins)
                statement.append_from(eager_joins, _copy_collection=False)

            if order_by:
                statement.append_order_by(*sql_util.ClauseAdapter(s3).copy_and_process(order_by))

            statement.append_order_by(*context.eager_order_by)
        else:
            if adapter:
                # TODO: make usage of the ClauseAdapter here to create row adapter, list
                # of primary columns
                context.primary_columns = [from_obj.corresponding_column(c) or c for c in context.primary_columns]
                context.row_adapter = mapperutil.create_row_adapter(from_obj, self.table)

            if self._distinct:
                if order_by:
                    order_by = [expression._literal_as_text(o) for o in util.to_list(order_by) or []]

                if self._distinct and order_by:
                    cf = util.Set()
                    for o in order_by:
                        cf.update(sql_util.find_columns(o))
                    for c in cf:
                        context.primary_columns.append(c)

            statement = sql.select(context.primary_columns + context.secondary_columns, whereclause, from_obj=from_obj, use_labels=True, for_update=for_update, order_by=util.to_list(order_by), **self._select_args())
            
            if context.eager_joins:
                if adapter:
                    context.eager_joins = adapter.traverse(context.eager_joins)
                statement.append_from(context.eager_joins, _copy_collection=False)

            if context.eager_order_by:
                if adapter:
                    context.eager_order_by = adapter.copy_and_process(context.eager_order_by)
                statement.append_order_by(*context.eager_order_by)

        context.statement = statement

        return context

    def _select_args(self):
        """Return a dictionary of attributes that can be applied to a ``sql.Select`` statement.
        """
        return {'limit':self._limit, 'offset':self._offset, 'distinct':self._distinct, 'group_by':self._group_by or None, 'having':self._having or None}


    def _get_entity_clauses(self, m):
        """for tuples added via add_entity() or add_column(), attempt to locate
        an AliasedClauses object which should be used to formulate the query as well
        as to process result rows."""

        (m, alias, alias_id) = m
        if alias is not None:
            return alias
        if alias_id is not None:
            try:
                return self._alias_ids[alias_id]
            except KeyError:
                raise exceptions.InvalidRequestError("Query has no alias identified by '%s'" % alias_id)
                
        if isinstance(m, type):
            m = mapper.class_mapper(m)
        if isinstance(m, mapper.Mapper):
            l = self._alias_ids.get(m)
            if l:
                if len(l) > 1:
                    raise exceptions.InvalidRequestError("Ambiguous join for entity '%s'; specify id=<someid> to query.join()/query.add_entity()" % str(m))
                else:
                    return l[0]
            else:
                return None
        elif isinstance(m, sql.ColumnElement):
            aliases = []
            for table in sql_util.find_tables(m, check_columns=True):
                for a in self._alias_ids.get(table, []):
                    aliases.append(a)
            if len(aliases) > 1:
                raise exceptions.InvalidRequestError("Ambiguous join for entity '%s'; specify id=<someid> to query.join()/query.add_column()" % str(m))
            elif len(aliases) == 1:
                return aliases[0]
            else:
                return None

    def __log_debug(self, msg):
        self.logger.debug(msg)

    def __str__(self):
        return str(self.compile())

    # DEPRECATED LAND !

    def list(self): #pragma: no cover
        """DEPRECATED.  use all()"""

        return list(self)

    def scalar(self): #pragma: no cover
        """DEPRECATED.  use first()"""

        return self.first()

    def _legacy_filter_by(self, *args, **kwargs): #pragma: no cover
        return self.filter(self._legacy_join_by(args, kwargs, start=self._joinpoint))

    def count_by(self, *args, **params): #pragma: no cover
        """DEPRECATED.  use query.filter_by(\**params).count()"""

        return self.count(self.join_by(*args, **params))


    def select_whereclause(self, whereclause=None, params=None, **kwargs): #pragma: no cover
        """DEPRECATED.  use query.filter(whereclause).all()"""

        q = self.filter(whereclause)._legacy_select_kwargs(**kwargs)
        if params is not None:
            q = q.params(params)
        return list(q)

    def _legacy_select_from(self, from_obj):
        q = self._clone()
        if len(from_obj) > 1:
            raise exceptions.ArgumentError("Multiple-entry from_obj parameter no longer supported")
        q._from_obj = from_obj[0]
        return q

    def _legacy_select_kwargs(self, **kwargs): #pragma: no cover
        q = self
        if "order_by" in kwargs and kwargs['order_by']:
            q = q.order_by(kwargs['order_by'])
        if "group_by" in kwargs:
            q = q.group_by(kwargs['group_by'])
        if "from_obj" in kwargs:
            q = q._legacy_select_from(kwargs['from_obj'])
        if "lockmode" in kwargs:
            q = q.with_lockmode(kwargs['lockmode'])
        if "distinct" in kwargs:
            q = q.distinct()
        if "limit" in kwargs:
            q = q.limit(kwargs['limit'])
        if "offset" in kwargs:
            q = q.offset(kwargs['offset'])
        return q


    def get_by(self, *args, **params): #pragma: no cover
        """DEPRECATED.  use query.filter_by(\**params).first()"""

        ret = self._extension.get_by(self, *args, **params)
        if ret is not mapper.EXT_CONTINUE:
            return ret

        return self._legacy_filter_by(*args, **params).first()

    def select_by(self, *args, **params): #pragma: no cover
        """DEPRECATED. use use query.filter_by(\**params).all()."""

        ret = self._extension.select_by(self, *args, **params)
        if ret is not mapper.EXT_CONTINUE:
            return ret

        return self._legacy_filter_by(*args, **params).list()

    def join_by(self, *args, **params): #pragma: no cover
        """DEPRECATED. use join() to construct joins based on attribute names."""

        return self._legacy_join_by(args, params, start=self._joinpoint)

    def _build_select(self, arg=None, params=None, **kwargs): #pragma: no cover
        if isinstance(arg, sql.FromClause) and arg.supports_execution():
            return self.from_statement(arg)
        else:
            return self.filter(arg)._legacy_select_kwargs(**kwargs)

    def selectfirst(self, arg=None, **kwargs): #pragma: no cover
        """DEPRECATED.  use query.filter(whereclause).first()"""

        return self._build_select(arg, **kwargs).first()

    def selectone(self, arg=None, **kwargs): #pragma: no cover
        """DEPRECATED.  use query.filter(whereclause).one()"""

        return self._build_select(arg, **kwargs).one()

    def select(self, arg=None, **kwargs): #pragma: no cover
        """DEPRECATED.  use query.filter(whereclause).all(), or query.from_statement(statement).all()"""

        ret = self._extension.select(self, arg=arg, **kwargs)
        if ret is not mapper.EXT_CONTINUE:
            return ret
        return self._build_select(arg, **kwargs).all()

    def execute(self, clauseelement, params=None, *args, **kwargs): #pragma: no cover
        """DEPRECATED.  use query.from_statement().all()"""

        return self._select_statement(clauseelement, params, **kwargs)

    def select_statement(self, statement, **params): #pragma: no cover
        """DEPRECATED.  Use query.from_statement(statement)"""

        return self._select_statement(statement, params)

    def select_text(self, text, **params): #pragma: no cover
        """DEPRECATED.  Use query.from_statement(statement)"""

        return self._select_statement(text, params)

    def _select_statement(self, statement, params=None, **kwargs): #pragma: no cover
        q = self.from_statement(statement)
        if params is not None:
            q = q.params(params)
        q._select_context_options(**kwargs)
        return list(q)

    def _select_context_options(self, populate_existing=None, version_check=None, only_load_props=None, refresh_instance=None): #pragma: no cover
        if populate_existing:
            self._populate_existing = populate_existing
        if version_check:
            self._version_check = version_check
        if refresh_instance:
            self._refresh_instance = refresh_instance
        if only_load_props:
            self._only_load_props = util.Set(only_load_props)
        return self

    def join_to(self, key): #pragma: no cover
        """DEPRECATED. use join() to create joins based on property names."""

        [keys, p] = self._locate_prop(key)
        return self.join_via(keys)

    def join_via(self, keys): #pragma: no cover
        """DEPRECATED. use join() to create joins based on property names."""

        mapper = self._joinpoint
        clause = None
        for key in keys:
            prop = mapper.get_property(key, resolve_synonyms=True)
            if clause is None:
                clause = prop.get_join(mapper)
            else:
                clause &= prop.get_join(mapper)
            mapper = prop.mapper

        return clause

    def _legacy_join_by(self, args, params, start=None): #pragma: no cover
        import properties

        clause = None
        for arg in args:
            if clause is None:
                clause = arg
            else:
                clause &= arg

        for key, value in params.iteritems():
            (keys, prop) = self._locate_prop(key, start=start)
            if isinstance(prop, properties.PropertyLoader):
                c = prop.compare(operators.eq, value) & self.join_via(keys[:-1])
            else:
                c = prop.compare(operators.eq, value) & self.join_via(keys)
            if clause is None:
                clause =  c
            else:
                clause &= c
        return clause

    def _locate_prop(self, key, start=None): #pragma: no cover
        import properties
        keys = []
        seen = util.Set()
        def search_for_prop(mapper_):
            if mapper_ in seen:
                return None
            seen.add(mapper_)

            prop = mapper_.get_property(key, resolve_synonyms=True, raiseerr=False)
            if prop is not None:
                if isinstance(prop, properties.PropertyLoader):
                    keys.insert(0, prop.key)
                return prop
            else:
                for prop in mapper_.iterate_properties:
                    if not isinstance(prop, properties.PropertyLoader):
                        continue
                    x = search_for_prop(prop.mapper)
                    if x:
                        keys.insert(0, prop.key)
                        return x
                else:
                    return None
        p = search_for_prop(start or self.mapper)
        if p is None:
            raise exceptions.InvalidRequestError("Can't locate property named '%s'" % key)
        return [keys, p]

    def selectfirst_by(self, *args, **params): #pragma: no cover
        """DEPRECATED. Use query.filter_by(\**kwargs).first()"""

        return self._legacy_filter_by(*args, **params).first()

    def selectone_by(self, *args, **params): #pragma: no cover
        """DEPRECATED. Use query.filter_by(\**kwargs).one()"""

        return self._legacy_filter_by(*args, **params).one()


for deprecated_method in ('list', 'scalar', 'count_by',
                          'select_whereclause', 'get_by', 'select_by',
                          'join_by', 'selectfirst', 'selectone', 'select',
                          'execute', 'select_statement', 'select_text',
                          'join_to', 'join_via', 'selectfirst_by',
                          'selectone_by'):
    setattr(Query, deprecated_method,
            util.deprecated(getattr(Query, deprecated_method),
                            add_deprecation_to_docstring=False))

Query.logger = logging.class_logger(Query)

class QueryContext(object):
    def __init__(self, query):
        self.query = query
        self.mapper = query.mapper
        self.session = query.session
        self.extension = query._extension
        self.statement = None
        self.row_adapter = None
        self.populate_existing = query._populate_existing
        self.version_check = query._version_check
        self.only_load_props = query._only_load_props
        self.refresh_instance = query._refresh_instance
        self.path = ()
        self.primary_columns = []
        self.secondary_columns = []
        self.eager_order_by = []
        self.eager_joins = None
        self.options = query._with_options
        self.attributes = query._attributes.copy()

    def exec_with_path(self, mapper, propkey, func, *args, **kwargs):
        oldpath = self.path
        self.path += (mapper.base_mapper, propkey)
        try:
            return func(*args, **kwargs)
        finally:
            self.path = oldpath

_runid = 1L
_id_lock = util.threading.Lock()

def _new_runid():
    global _runid
    _id_lock.acquire()
    try:
        _runid += 1
        return _runid
    finally:
        _id_lock.release()