replace
)
- def _entity_zero(self):
+ def _query_entity_zero(self):
+ """Return the first QueryEntity."""
return self._entities[0]
def _mapper_zero(self):
- # TODO: self._select_from_entity is not a mapper
- # so this method is misnamed
+ """return the Mapper associated with the first QueryEntity."""
+ return self._entities[0].mapper
+
+ def _entity_zero(self):
+ """Return the 'entity' (mapper or AliasedClass) associated
+ with the first QueryEntity, or alternatively the 'select from'
+ entity if specified."""
+
return self._select_from_entity \
if self._select_from_entity is not None \
- else self._entity_zero().entity_zero
+ else self._query_entity_zero().entity_zero
@property
def _mapper_entities(self):
def _joinpoint_zero(self):
return self._joinpoint.get(
'_joinpoint_entity',
- self._mapper_zero()
+ self._entity_zero()
)
def _bind_mapper(self):
- ezero = self._mapper_zero()
+ ezero = self._entity_zero()
if ezero is not None:
insp = inspect(ezero)
if not insp.is_clause_element:
return None
- 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."
- )
- return self._mapper_zero()
-
def _only_full_mapper_zero(self, methname):
if self._entities != [self._primary_entity]:
raise sa_exc.InvalidRequestError(
"""
if property is None:
- mapper_zero = inspect(self._mapper_zero()).mapper
+ mapper_zero = self._mapper_zero()
mapper = object_mapper(instance)
statement.correlate(None)
q = self._from_selectable(fromclause)
q._enable_single_crit = False
- q._select_from_entity = self._mapper_zero()
+ q._select_from_entity = self._entity_zero()
if entities:
q._set_entities(entities)
return q
from sqlalchemy import testing
from sqlalchemy.testing import (
- fixtures, eq_, assert_raises, assert_raises_message, AssertsCompiledSQL)
+ fixtures, eq_, is_, assert_raises,
+ assert_raises_message, AssertsCompiledSQL)
from sqlalchemy import (
exc as sa_exc, util, Integer, Table, String, ForeignKey, select, func,
and_, asc, desc, inspect, literal_column, cast, exists, text)
"ON users.id = addresses_1.user_id"),
]:
q = s.query(crit)
- mzero = q._mapper_zero()
- assert mzero.mapped_table is q._entity_zero().selectable
+ mzero = q._entity_zero()
+ is_(mzero.mapped_table, q._query_entity_zero().selectable)
q = q.join(j)
self.assert_compile(q, exp)
"ON users_1.id = addresses_1.user_id")
]:
q = s.query(crit)
- mzero = q._mapper_zero()
- assert inspect(mzero).selectable is q._entity_zero().selectable
+ mzero = q._entity_zero()
+ is_(inspect(mzero).selectable, q._query_entity_zero().selectable)
q = q.join(j)
self.assert_compile(q, exp)