return generate
class Query(object):
- """Encapsulates the object-fetching operations provided by Mappers."""
-
+ """ORM-level SQL construction object."""
+
+ _enable_eagerloads = True
+ _enable_assertions = True
+ _with_labels = False
+ _criterion = None
+ _yield_per = None
+ _lockmode = None
+ _order_by = False
+ _group_by = False
+ _having = None
+ _distinct = False
+ _offset = None
+ _limit = None
+ _statement = None
+ _joinpoint = None
+ _correlate = frozenset()
+ _populate_existing = False
+ _version_check = False
+ _autoflush = True
+ _current_path = ()
+ _only_load_props = None
+ _refresh_state = None
+ _from_obj = ()
+ _filter_aliases = None
+ _from_obj_alias = None
+ _currenttables = frozenset()
+
def __init__(self, entities, session=None):
self.session = session
self._with_options = []
- self._lockmode = None
- 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._correlate = set()
- self._joinpoint = None
- self._with_labels = False
- self._enable_eagerloads = True
- self.__joinable_tables = None
- self._having = None
- self._populate_existing = False
- self._version_check = False
- self._autoflush = True
self._attributes = {}
- self._current_path = ()
- self._only_load_props = None
- self._refresh_state = None
- self._from_obj = ()
self._polymorphic_adapters = {}
- self._filter_aliases = None
- self._from_obj_alias = None
- self.__currenttables = set()
self._set_entities(entities)
def _set_entities(self, entities, entity_wrapper=None):
for ent in util.to_list(entities):
entity_wrapper(self, ent)
- self.__setup_aliasizers(self._entities)
+ self._setup_aliasizers(self._entities)
- def __setup_aliasizers(self, entities):
+ def _setup_aliasizers(self, entities):
if hasattr(self, '_mapper_adapter_map'):
# usually safe to share a single map, but copying to prevent
# subtle leaks if end-user is reusing base query with arbitrary
mapper, selectable, is_aliased_class = _entity_info(entity)
if not is_aliased_class and mapper.with_polymorphic:
with_polymorphic = mapper._with_polymorphic_mappers
- self.__mapper_loads_polymorphically_with(mapper, sql_util.ColumnAdapter(selectable, mapper._equivalent_columns))
+ self.__mapper_loads_polymorphically_with(mapper,
+ sql_util.ColumnAdapter(selectable, mapper._equivalent_columns))
adapter = None
elif is_aliased_class:
adapter = sql_util.ColumnAdapter(selectable, mapper._equivalent_columns)
for m in m2.iterate_to_root():
self._polymorphic_adapters[m.mapped_table] = self._polymorphic_adapters[m.local_table] = adapter
- def __set_select_from(self, from_obj):
+ def _set_select_from(self, from_obj):
if isinstance(from_obj, expression._SelectBaseMixin):
from_obj = from_obj.alias()
self._from_obj_alias = sql_util.ColumnAdapter(from_obj, equivs)
def _get_polymorphic_adapter(self, entity, selectable):
- self.__mapper_loads_polymorphically_with(entity.mapper, sql_util.ColumnAdapter(selectable, entity.mapper._equivalent_columns))
+ self.__mapper_loads_polymorphically_with(entity.mapper,
+ sql_util.ColumnAdapter(selectable, entity.mapper._equivalent_columns))
def _reset_polymorphic_adapter(self, mapper):
for m2 in mapper._with_polymorphic_mappers:
self._polymorphic_adapters.pop(m.mapped_table, None)
self._polymorphic_adapters.pop(m.local_table, None)
- def __reset_joinpoint(self):
+ def _reset_joinpoint(self):
self._joinpoint = None
self._filter_aliases = None
return clause
if getattr(self, '_disable_orm_filtering', not orm_only):
- return visitors.replacement_traverse(clause, {'column_collections':False}, self.__replace_element(adapters))
+ return visitors.replacement_traverse(
+ clause,
+ {'column_collections':False},
+ self.__replace_element(adapters)
+ )
else:
- return visitors.replacement_traverse(clause, {'column_collections':False}, self.__replace_orm_element(adapters))
+ return visitors.replacement_traverse(
+ clause,
+ {'column_collections':False},
+ self.__replace_orm_element(adapters)
+ )
def _entity_zero(self):
return self._entities[0]
def _only_mapper_zero(self, rationale=None):
if len(self._entities) > 1:
- raise sa_exc.InvalidRequestError(rationale or "This operation requires a Query against a single mapper.")
+ raise sa_exc.InvalidRequestError(
+ rationale or "This operation requires a Query against a single mapper."
+ )
return self._mapper_zero()
def _only_entity_zero(self, rationale=None):
if len(self._entities) > 1:
- raise sa_exc.InvalidRequestError(rationale or "This operation requires a Query against a single mapper.")
+ raise sa_exc.InvalidRequestError(
+ rationale or "This operation requires a Query against a single mapper."
+ )
return self._entity_zero()
def _generate_mapper_zero(self):
equivs.update(ent.mapper._equivalent_columns)
return equivs
- def __no_criterion_condition(self, meth):
+ def _no_criterion_condition(self, meth):
+ if not self._enable_assertions:
+ return
if self._criterion or self._statement or self._from_obj or \
self._limit is not None or self._offset is not None or \
self._group_by:
self._from_obj = ()
self._statement = self._criterion = None
self._order_by = self._group_by = self._distinct = False
- self.__joined_tables = {}
- def __no_clauseelement_condition(self, meth):
+ def _no_clauseelement_condition(self, meth):
+ if not self._enable_assertions:
+ return
if self._order_by:
raise sa_exc.InvalidRequestError("Query.%s() being called on a Query with existing criterion. " % meth)
- self.__no_criterion_condition(meth)
+ self._no_criterion_condition(meth)
- def __no_statement_condition(self, meth):
+ def _no_statement_condition(self, meth):
+ if not self._enable_assertions:
+ return
if self._statement:
raise sa_exc.InvalidRequestError(
("Query.%s() being called on a Query with an existing full "
"statement - can't apply criterion.") % meth)
- def __no_limit_offset(self, meth):
+ def _no_limit_offset(self, meth):
+ if not self._enable_assertions:
+ return
if self._limit is not None or self._offset is not None:
- # TODO: do we want from_self() to be implicit here ? i vote explicit for the time being
- raise sa_exc.InvalidRequestError("Query.%s() being called on a Query which already has LIMIT or OFFSET applied. "
- "To modify the row-limited results of a Query, call from_self() first. Otherwise, call %s() before limit() or offset() are applied." % (meth, meth)
+ raise sa_exc.InvalidRequestError(
+ "Query.%s() being called on a Query which already has LIMIT or OFFSET applied. "
+ "To modify the row-limited results of a Query, call from_self() first. "
+ "Otherwise, call %s() before limit() or offset() are applied." % (meth, meth)
)
-
- def __get_options(self, populate_existing=None, version_check=None, only_load_props=None, refresh_state=None):
+ def _get_options(self, populate_existing=None,
+ version_check=None,
+ only_load_props=None,
+ refresh_state=None):
if populate_existing:
self._populate_existing = populate_existing
if version_check:
def statement(self):
"""The full SELECT statement represented by this Query."""
- return self._compile_context(labels=self._with_labels).statement._annotate({'_halt_adapt': True})
+ return self._compile_context(labels=self._with_labels).\
+ statement._annotate({'_halt_adapt': True})
def subquery(self):
"""return the full SELECT statement represented by this Query, embedded within an Alias.
"""
self._with_labels = True
-
+
+ @_generative()
+ def enable_assertions(self, value):
+ """Control whether assertions are generated.
+
+ When set to False, the returned Query will
+ not assert its state before certain operations,
+ including that LIMIT/OFFSET has not been applied
+ when filter() is called, no criterion exists
+ when get() is called, and no "from_statement()"
+ exists when filter()/order_by()/group_by() etc.
+ is called. This more permissive mode is used by
+ custom Query subclasses to specify criterion or
+ other modifiers outside of the usual usage patterns.
+
+ Care should be taken to ensure that the usage
+ pattern is even possible. A statement applied
+ by from_statement() will override any criterion
+ set by filter() or order_by(), for example.
+
+ """
+ self._enable_assertions = value
+
@property
def whereclause(self):
"""The WHERE criterion for this Query."""
"""
self._current_path = path
- @_generative(__no_clauseelement_condition)
+ @_generative(_no_clauseelement_condition)
def with_polymorphic(self, cls_or_mappers, selectable=None, discriminator=None):
"""Load columns for descendant mappers of this Query's mapper.
if hasattr(ident, '__composite_values__'):
ident = ident.__composite_values__()
- key = self._only_mapper_zero("get() can only be used against a single mapped class.").identity_key_from_primary_key(ident)
+ key = self._only_mapper_zero(
+ "get() can only be used against a single mapped class."
+ ).identity_key_from_primary_key(ident)
return self._get(key, ident)
@classmethod
if isinstance(prop, properties.PropertyLoader) and prop.mapper is self._mapper_zero():
break
else:
- raise sa_exc.InvalidRequestError("Could not locate a property which relates instances of class '%s' to instances of class '%s'" % (self._mapper_zero().class_.__name__, instance.__class__.__name__))
+ raise sa_exc.InvalidRequestError(
+ "Could not locate a property which relates instances "
+ "of class '%s' to instances of class '%s'" %
+ (self._mapper_zero().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))
self._entities = list(self._entities)
m = _MapperEntity(self, entity)
- self.__setup_aliasizers([m])
+ self._setup_aliasizers([m])
def from_self(self, *entities):
"""return a Query that selects from this Query's SELECT statement.
self._statement = self._criterion = None
self._order_by = self._group_by = self._distinct = False
self._limit = self._offset = None
- self.__set_select_from(fromclause)
+ self._set_select_from(fromclause)
def values(self, *columns):
"""Return an iterator yielding result tuples corresponding to the given list of columns"""
_ColumnEntity(self, column)
# _ColumnEntity may add many entities if the
# given arg is a FROM clause
- self.__setup_aliasizers(self._entities[l:])
+ self._setup_aliasizers(self._entities[l:])
def options(self, *args):
"""Return a new Query object, applying the given list of
MapperOptions.
"""
- return self.__options(False, *args)
+ return self._options(False, *args)
def _conditional_options(self, *args):
- return self.__options(True, *args)
+ return self._options(True, *args)
@_generative()
- def __options(self, conditional, *args):
+ def _options(self, conditional, *args):
# most MapperOptions write to the '_attributes' dictionary,
# so copy that as well
self._attributes = self._attributes.copy()
self._params = self._params.copy()
self._params.update(kwargs)
- @_generative(__no_statement_condition, __no_limit_offset)
+ @_generative(_no_statement_condition, _no_limit_offset)
def filter(self, criterion):
"""apply the given filtering criterion to the query and return the newly resulting ``Query``
return self.filter(sql.and_(*clauses))
- @_generative(__no_statement_condition, __no_limit_offset)
+ @_generative(_no_statement_condition, _no_limit_offset)
@util.accepts_a_list_as_starargs(list_deprecation='pending')
def order_by(self, *criterion):
"""apply one or more ORDER BY criterion to the query and return the newly resulting ``Query``"""
else:
self._order_by = self._order_by + criterion
- @_generative(__no_statement_condition, __no_limit_offset)
+ @_generative(_no_statement_condition, _no_limit_offset)
@util.accepts_a_list_as_starargs(list_deprecation='pending')
def group_by(self, *criterion):
"""apply one or more GROUP BY criterion to the query and return the newly resulting ``Query``"""
else:
self._group_by = self._group_by + criterion
- @_generative(__no_statement_condition, __no_limit_offset)
+ @_generative(_no_statement_condition, _no_limit_offset)
def having(self, criterion):
"""apply a HAVING criterion to the query and return the newly resulting ``Query``."""
aliased, from_joinpoint = kwargs.pop('aliased', False), kwargs.pop('from_joinpoint', False)
if kwargs:
raise TypeError("unknown arguments: %s" % ','.join(kwargs.iterkeys()))
- return self.__join(props, outerjoin=False, create_aliases=aliased, from_joinpoint=from_joinpoint)
+ return self._join(props, outerjoin=False, create_aliases=aliased, from_joinpoint=from_joinpoint)
@util.accepts_a_list_as_starargs(list_deprecation='pending')
def outerjoin(self, *props, **kwargs):
aliased, from_joinpoint = kwargs.pop('aliased', False), kwargs.pop('from_joinpoint', False)
if kwargs:
raise TypeError("unknown arguments: %s" % ','.join(kwargs.iterkeys()))
- return self.__join(props, outerjoin=True, create_aliases=aliased, from_joinpoint=from_joinpoint)
+ return self._join(props, outerjoin=True, create_aliases=aliased, from_joinpoint=from_joinpoint)
- @_generative(__no_statement_condition, __no_limit_offset)
- def __join(self, keys, outerjoin, create_aliases, from_joinpoint):
+ @_generative(_no_statement_condition, _no_limit_offset)
+ def _join(self, keys, outerjoin, create_aliases, from_joinpoint):
# copy collections that may mutate so they do not affect
# the copied-from query.
- self.__currenttables = set(self.__currenttables)
+ self._currenttables = set(self._currenttables)
self._polymorphic_adapters = self._polymorphic_adapters.copy()
# start from the beginning unless from_joinpoint is set.
if not from_joinpoint:
- self.__reset_joinpoint()
+ self._reset_joinpoint()
clause = replace_clause_index = None
elif prop:
# for joins across plain relation()s, try not to specify the
- # same joins twice. the __currenttables collection tracks
+ # same joins twice. the _currenttables collection tracks
# what plain mapped tables we've joined to already.
- if prop.table in self.__currenttables:
- if prop.secondary is not None and prop.secondary not in self.__currenttables:
+ if prop.table in self._currenttables:
+ if prop.secondary is not None and prop.secondary not in self._currenttables:
# TODO: this check is not strong enough for different paths to the same endpoint which
# does not use secondary tables
raise sa_exc.InvalidRequestError("Can't join to property '%s'; a path to this "
continue
if prop.secondary:
- self.__currenttables.add(prop.secondary)
- self.__currenttables.add(prop.table)
+ self._currenttables.add(prop.secondary)
+ self._currenttables.add(prop.table)
if of_type:
right_entity = of_type
# future joins with from_joinpoint=True join from our established right_entity.
self._joinpoint = right_entity
- @_generative(__no_statement_condition)
+ @_generative(_no_statement_condition)
def reset_joinpoint(self):
"""return a new Query reset the 'joinpoint' of this Query reset
back to the starting mapper. Subsequent generative calls will
the root.
"""
- self.__reset_joinpoint()
+ self._reset_joinpoint()
- @_generative(__no_clauseelement_condition)
+ @_generative(_no_clauseelement_condition)
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_obj = from_obj[-1]
if not isinstance(from_obj, expression.FromClause):
raise sa_exc.ArgumentError("select_from() accepts FromClause objects only.")
- self.__set_select_from(from_obj)
+ self._set_select_from(from_obj)
def __getitem__(self, item):
if isinstance(item, slice):
else:
return list(self[item:item+1])[0]
- @_generative(__no_statement_condition)
+ @_generative(_no_statement_condition)
def slice(self, start, stop):
"""apply LIMIT/OFFSET to the ``Query`` based on a range and return the newly resulting ``Query``."""
if start is not None and stop is not None:
elif start is not None and stop is None:
self._offset = (self._offset or 0) + start
- @_generative(__no_statement_condition)
+ @_generative(_no_statement_condition)
def limit(self, limit):
"""Apply a ``LIMIT`` to the query and return the newly resulting
"""
self._limit = limit
- @_generative(__no_statement_condition)
+ @_generative(_no_statement_condition)
def offset(self, offset):
"""Apply an ``OFFSET`` to the query and return the newly resulting
``Query``.
"""
self._offset = offset
- @_generative(__no_statement_condition)
+ @_generative(_no_statement_condition)
def distinct(self):
"""Apply a ``DISTINCT`` to the query and return the newly resulting
``Query``.
"""
return list(self)
- @_generative(__no_clauseelement_condition)
+ @_generative(_no_clauseelement_condition)
def from_statement(self, statement):
"""Execute the given SELECT statement and return results.
if refresh_state is None:
q = self._clone()
- q.__no_criterion_condition("get")
+ q._no_criterion_condition("get")
else:
q = self._clone()
if lockmode is not None:
q._lockmode = lockmode
- q.__get_options(
+ q._get_options(
populate_existing=bool(refresh_state),
version_check=(lockmode is not None),
only_load_props=only_load_props,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
