query.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

# orm/query.py
# Copyright (C) 2005, 2006, 2007, 2008 Michael Bayer [e-mail]
#
# 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 propagated 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 desirable 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 desirable 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)