_offset = None
_limit = None
_statement = None
- _joinpoint = None
_correlate = frozenset()
_populate_existing = False
_version_check = False
_from_obj = ()
_filter_aliases = None
_from_obj_alias = None
- _currenttables = frozenset()
+ _joinpath = _joinpoint = {}
def __init__(self, entities, session=None):
self.session = session
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):
- if isinstance(from_obj, expression._SelectBaseMixin):
- from_obj = from_obj.alias()
-
- self._from_obj = (from_obj,)
- equivs = self.__all_equivs()
+ def _set_select_from(self, *obj):
+
+ fa = []
+ for from_obj in obj:
+ if isinstance(from_obj, expression._SelectBaseMixin):
+ from_obj = from_obj.alias()
+ fa.append(from_obj)
- if isinstance(from_obj, expression.Alias):
- self._from_obj_alias = sql_util.ColumnAdapter(from_obj, equivs)
+ self._from_obj = tuple(fa)
+ # TODO: only use this adapter for from_self() ? right
+ # now its usage is somewhat arbitrary.
+ if len(self._from_obj) == 1 and isinstance(self._from_obj[0], expression.Alias):
+ equivs = self.__all_equivs()
+ self._from_obj_alias = sql_util.ColumnAdapter(self._from_obj[0], equivs)
+
def _get_polymorphic_adapter(self, entity, selectable):
self.__mapper_loads_polymorphically_with(entity.mapper,
sql_util.ColumnAdapter(selectable, entity.mapper._equivalent_columns))
self._polymorphic_adapters.pop(m.mapped_table, None)
self._polymorphic_adapters.pop(m.local_table, None)
- def _reset_joinpoint(self):
- self._joinpoint = None
- self._filter_aliases = None
-
def __adapt_polymorphic_element(self, element):
if isinstance(element, expression.FromClause):
search = element
def _adapt_clause(self, clause, as_filter, orm_only):
adapters = []
if as_filter and self._filter_aliases:
- adapters.append(self._filter_aliases.replace)
+ for fa in self._filter_aliases._visitor_iterator:
+ adapters.append(fa.replace)
if self._from_obj_alias:
adapters.append(self._from_obj_alias.replace)
yield ent
def _joinpoint_zero(self):
- return self._joinpoint or self._entity_zero().entity_zero
+ return self._joinpoint.get('_joinpoint_entity', self._entity_zero().entity_zero)
def _mapper_zero_or_none(self):
if not getattr(self._entities[0], 'primary_entity', False):
@_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)
+ """consumes arguments from join() or outerjoin(), places them into a consistent
+ format with which to form the actual JOIN constructs.
+
+ """
self._polymorphic_adapters = self._polymorphic_adapters.copy()
- # start from the beginning unless from_joinpoint is set.
if not from_joinpoint:
self._reset_joinpoint()
- clause = replace_clause_index = None
-
- # after the method completes,
- # the query's joinpoint will be set to this.
- right_entity = None
-
for arg1 in util.to_list(keys):
- aliased_entity = False
- alias_criterion = False
- left_entity = right_entity
- prop = of_type = right_entity = right_mapper = None
-
- # distinguish between tuples, scalar args
if isinstance(arg1, tuple):
arg1, arg2 = arg1
else:
# is a little bit of legacy behavior still at work here
# which means they might be in either order. may possibly
# lock this down to (right_entity, onclause) in 0.6.
- if isinstance(arg2, (interfaces.PropComparator, basestring)):
- onclause = arg2
- right_entity = arg1
- elif isinstance(arg1, (interfaces.PropComparator, basestring)):
- onclause = arg1
- right_entity = arg2
+ if isinstance(arg1, (interfaces.PropComparator, basestring)):
+ right_entity, onclause = arg2, arg1
else:
- onclause = arg2
- right_entity = arg1
+ right_entity, onclause = arg1, arg2
+
+ left_entity = prop = None
+
+ if isinstance(onclause, basestring):
+ left_entity = self._joinpoint_zero()
+
+ descriptor, prop = _entity_descriptor(left_entity, onclause)
+ onclause = descriptor
- # extract info from the onclause argument, determine
- # left_entity and right_entity.
if isinstance(onclause, interfaces.PropComparator):
- of_type = getattr(onclause, '_of_type', None)
+ if not right_entity:
+ right_entity = onclause.property.mapper
+ of_type = getattr(onclause, '_of_type', None)
+ if of_type:
+ right_entity = of_type
+ else:
+ right_entity = onclause.property.mapper
+
+ left_entity = onclause.parententity
+
prop = onclause.property
- descriptor = onclause
+ if not isinstance(onclause, attributes.QueryableAttribute):
+ onclause = prop
+
+ if not create_aliases:
+ # check for this path already present.
+ # don't render in that case.
+ if (left_entity, right_entity, prop.key) in self._joinpoint:
+ self._joinpoint = self._joinpoint[(left_entity, right_entity, prop.key)]
+ continue
- if not left_entity:
- left_entity = onclause.parententity
+ elif onclause is not None and right_entity is None:
+ # TODO: no coverage here
+ raise NotImplementedError("query.join(a==b) not supported.")
+
+ self._join_left_to_right(left_entity, right_entity, onclause, outerjoin, create_aliases, prop)
- if of_type:
- right_mapper = of_type
- else:
- right_mapper = prop.mapper
+ def _join_left_to_right(self, left, right, onclause, outerjoin, create_aliases, prop):
+ """append a JOIN to the query's from clause."""
+
+ if left is None:
+ left = self._joinpoint_zero()
- if not right_entity:
- right_entity = right_mapper
+ left_mapper, left_selectable, left_is_aliased = _entity_info(left)
+ right_mapper, right_selectable, is_aliased_class = _entity_info(right)
- elif isinstance(onclause, basestring):
- if not left_entity:
- left_entity = self._joinpoint_zero()
+ if right_mapper and prop and not right_mapper.common_parent(prop.mapper):
+ raise sa_exc.InvalidRequestError(
+ "Join target %s does not correspond to the right side of join condition %s" % (right, onclause)
+ )
- descriptor, prop = _entity_descriptor(left_entity, onclause)
- right_mapper = prop.mapper
+ if not right_mapper and prop:
+ right_mapper = prop.mapper
- if not right_entity:
- right_entity = right_mapper
- elif not left_entity:
- left_entity = self._joinpoint_zero()
+ need_adapter = False
- if not clause and self._from_obj:
- mp, left_selectable, is_aliased_class = _entity_info(left_entity)
-
- replace_clause_index, clause = sql_util.find_join_source(self._from_obj, left_selectable)
- if not clause:
- clause = left_selectable
-
- if not clause and left_entity:
- for ent in self._entities:
- if ent.corresponds_to(left_entity):
- clause = ent.selectable
- break
-
- # TODO:
- # this provides one kind of "backwards join"
- # tested in test/orm/query.py.
- # removal of this has been considered, but maybe not
- # see [ticket:1445]
- if not clause:
- if isinstance(onclause, interfaces.PropComparator):
- clause = onclause.__clause_element__()
-
- if not clause:
- raise sa_exc.InvalidRequestError("Could not find a FROM clause to join from")
-
- # if we have a MapperProperty and the onclause is not already
- # an instrumented descriptor. this catches of_type()
- # PropComparators and string-based on clauses.
- if prop and not isinstance(onclause, attributes.QueryableAttribute):
- onclause = prop
-
- # start looking at the right side of the join
-
- mp, right_selectable, is_aliased_class = _entity_info(right_entity)
-
- if mp is not None and right_mapper is not None and not mp.common_parent(right_mapper):
+ if right_mapper and right is right_selectable:
+ if not right_selectable.is_derived_from(right_mapper.mapped_table):
raise sa_exc.InvalidRequestError(
- "Join target %s does not correspond to the right side of join condition %s" % (right_entity, onclause)
- )
+ "Selectable '%s' is not derived from '%s'" %
+ (right_selectable.description, right_mapper.mapped_table.description))
- if not right_mapper and mp:
- right_mapper = mp
-
- # determine if we need to wrap the right hand side in an alias.
- # this occurs based on the create_aliases flag, or if the target
- # is a selectable, Join, or polymorphically-loading mapper
- if right_mapper and not is_aliased_class:
- if right_entity is right_selectable:
-
- if not right_selectable.is_derived_from(right_mapper.mapped_table):
- raise sa_exc.InvalidRequestError(
- "Selectable '%s' is not derived from '%s'" %
- (right_selectable.description, right_mapper.mapped_table.description))
-
- if not isinstance(right_selectable, expression.Alias):
- right_selectable = right_selectable.alias()
-
- right_entity = aliased(right_mapper, right_selectable)
- alias_criterion = True
-
- elif create_aliases:
- right_entity = aliased(right_mapper)
- alias_criterion = True
-
- elif right_mapper.with_polymorphic or isinstance(right_mapper.mapped_table, expression.Join):
- right_entity = aliased(right_mapper)
- alias_criterion = True
- aliased_entity = True
-
- elif prop:
- # for joins across plain relation()s, try not to specify the
- # 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:
- # 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 "
- "table along a different secondary table already "
- "exists. Use the `alias=True` argument to `join()`." % descriptor)
- continue
+ if not isinstance(right_selectable, expression.Alias):
+ right_selectable = right_selectable.alias()
- if prop.secondary:
- self._currenttables.add(prop.secondary)
- self._currenttables.add(prop.table)
+ right = aliased(right_mapper, right_selectable)
+ need_adapter = True
+
+ aliased_entity = right_mapper and \
+ not is_aliased_class and \
+ (
+ right_mapper.with_polymorphic or
+ isinstance(right_mapper.mapped_table, expression.Join)
+ )
+
+ if not need_adapter and (create_aliases or aliased_entity):
+ right = aliased(right)
+ need_adapter = True
+
+ # if joining on a MapperProperty path,
+ # track the path to prevent redundant joins
+ if not create_aliases and prop:
+
+ self._joinpoint = jp = {
+ '_joinpoint_entity':right,
+ 'prev':((left, right, prop.key), self._joinpoint)
+ }
+
+ # copy backwards to the root of the _joinpath
+ # dict, so that no existing dict in the path is mutated
+ while 'prev' in jp:
+ f, prev = jp['prev']
+ prev = prev.copy()
+ prev[f] = jp
+ jp['prev'] = (f, prev)
+ jp = prev
+
+ self._joinpath = jp
- if of_type:
- right_entity = of_type
- else:
- right_entity = prop.mapper
-
- # create adapters to the right side, if we've created aliases
- if alias_criterion:
- right_adapter = ORMAdapter(right_entity,
- equivalents=right_mapper._equivalent_columns, chain_to=self._filter_aliases)
-
- # if the onclause is a ClauseElement, adapt it with our right
- # adapter, then with our query-wide adaptation if any.
- if isinstance(onclause, expression.ClauseElement):
- if alias_criterion:
- onclause = right_adapter.traverse(onclause)
- onclause = self._adapt_clause(onclause, False, True)
-
- # determine if we want _ORMJoin to alias the onclause
- # to the given left side. This is used if we're joining against a
- # select_from() selectable, from_self() call, or the onclause
- # has been resolved into a MapperProperty. Otherwise we assume
- # the onclause itself contains more specific information on how to
- # construct the onclause.
- join_to_left = not is_aliased_class or \
- onclause is prop or \
- self._from_obj_alias and clause is self._from_obj[0]
-
- # create the join
- clause = orm_join(clause, right_entity, onclause, isouter=outerjoin, join_to_left=join_to_left)
-
- # set up state for the query as a whole
- if alias_criterion:
- # adapt filter() calls based on our right side adaptation
- self._filter_aliases = right_adapter
-
- # if a polymorphic entity was aliased, establish that
- # so that MapperEntity/ColumnEntity can pick up on it
- # and adapt when it renders columns and fetches them from results
- if aliased_entity:
- self.__mapper_loads_polymorphically_with(
- right_mapper,
- ORMAdapter(right_entity, equivalents=right_mapper._equivalent_columns)
- )
-
- if replace_clause_index is not None:
- l = list(self._from_obj)
- l[replace_clause_index] = clause
- self._from_obj = tuple(l)
else:
- self._from_obj = self._from_obj + (clause,)
+ self._joinpoint = {
+ '_joinpoint_entity':right
+ }
+
+ # if an alias() of the right side was generated here,
+ # apply an adapter to all subsequent filter() calls
+ # until reset_joinpoint() is called.
+ if need_adapter:
+ self._filter_aliases = ORMAdapter(right,
+ equivalents=right_mapper._equivalent_columns, chain_to=self._filter_aliases)
+
+ # if the onclause is a ClauseElement, adapt it with any
+ # adapters that are in place right now
+ if isinstance(onclause, expression.ClauseElement):
+ onclause = self._adapt_clause(onclause, True, True)
+
+ # if an alias() on the right side was generated,
+ # which is intended to wrap a the right side in a subquery,
+ # ensure that columns retrieved from this target in the result
+ # set are also adapted.
+ if aliased_entity:
+ self.__mapper_loads_polymorphically_with(
+ right_mapper,
+ ORMAdapter(
+ right,
+ equivalents=right_mapper._equivalent_columns
+ )
+ )
+
+ join_to_left = not is_aliased_class
+
+ if self._from_obj:
+ replace_clause_index, clause = sql_util.find_join_source(self._from_obj, left_selectable)
+ if clause:
+ # the entire query's FROM clause is an alias of itself (i.e. from_self(), similar).
+ # if the left clause is that one, ensure it aliases to the left side.
+ if self._from_obj_alias and clause is self._from_obj[0]:
+ join_to_left = True
+
+ clause = orm_join(clause, right, onclause, isouter=outerjoin, join_to_left=join_to_left)
+
+ self._from_obj = \
+ self._from_obj[:replace_clause_index] + \
+ (clause, ) + \
+ self._from_obj[replace_clause_index + 1:]
+ return
+
+ if left_mapper:
+ for ent in self._entities:
+ if ent.corresponds_to(left):
+ clause = ent.selectable
+ break
+ else:
+ clause = left
+ else:
+ clause = None
- # future joins with from_joinpoint=True join from our established right_entity.
- self._joinpoint = right_entity
+ if clause is None:
+ raise sa_exc.InvalidRequestError("Could not find a FROM clause to join from")
+
+ clause = orm_join(clause, right, onclause, isouter=outerjoin, join_to_left=join_to_left)
+ self._from_obj = self._from_obj + (clause,)
+
+ def _reset_joinpoint(self):
+ self._joinpoint = self._joinpath
+ self._filter_aliases = None
@_generative(_no_statement_condition)
def reset_joinpoint(self):
self._reset_joinpoint()
@_generative(_no_clauseelement_condition)
- def select_from(self, from_obj):
+ 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.
-
"""
- if isinstance(from_obj, (tuple, list)):
- # from_obj is actually a list again as of 0.5.3. so this restriction here
- # is somewhat artificial, but is still in place since select_from() implies aliasing all further
- # criterion against what's placed here, and its less complex to only
- # keep track of a single aliased FROM element being selected against. This could in theory be opened
- # up again to more complexity.
- 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 not isinstance(from_obj, expression.FromClause):
- raise sa_exc.ArgumentError("select_from() accepts FromClause objects only.")
- self._set_select_from(from_obj)
+ for fo in from_obj:
+ if not isinstance(fo, expression.FromClause):
+ raise sa_exc.ArgumentError("select_from() accepts FromClause objects only.")
+
+ self._set_select_from(*from_obj)
def __getitem__(self, item):
if isinstance(item, slice):
k = sess.query(Keyword).with_parent(i1).all()
assert [Keyword(name='red'), Keyword(name='small'), Keyword(name='square')] == k
+class InheritedJoinTest(_base.MappedTest, AssertsCompiledSQL):
+ run_setup_mappers = 'once'
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('companies', metadata,
+ Column('company_id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)))
+
+ Table('people', metadata,
+ Column('person_id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('company_id', Integer, ForeignKey('companies.company_id')),
+ Column('name', String(50)),
+ Column('type', String(30)))
+
+ Table('engineers', metadata,
+ Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+ Column('status', String(30)),
+ Column('engineer_name', String(50)),
+ Column('primary_language', String(50)),
+ )
+
+ Table('machines', metadata,
+ Column('machine_id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)),
+ Column('engineer_id', Integer, ForeignKey('engineers.person_id')))
+
+ Table('managers', metadata,
+ Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+ Column('status', String(30)),
+ Column('manager_name', String(50))
+ )
+
+ Table('boss', metadata,
+ Column('boss_id', Integer, ForeignKey('managers.person_id'), primary_key=True),
+ Column('golf_swing', String(30)),
+ )
+
+ Table('paperwork', metadata,
+ Column('paperwork_id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('description', String(50)),
+ Column('person_id', Integer, ForeignKey('people.person_id')))
+
+ @classmethod
+ @testing.resolve_artifact_names
+ def setup_classes(cls):
+ class Company(_fixtures.Base):
+ pass
+ class Person(_fixtures.Base):
+ pass
+ class Engineer(Person):
+ pass
+ class Manager(Person):
+ pass
+ class Boss(Manager):
+ pass
+ class Machine(_fixtures.Base):
+ pass
+ class Paperwork(_fixtures.Base):
+ pass
+
+ mapper(Company, companies, properties={
+ 'employees':relation(Person, order_by=people.c.person_id)
+ })
+
+ mapper(Machine, machines)
+
+ mapper(Person, people,
+ polymorphic_on=people.c.type, polymorphic_identity='person', order_by=people.c.person_id,
+ properties={
+ 'paperwork':relation(Paperwork, order_by=paperwork.c.paperwork_id)
+ })
+ mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer', properties={
+ 'machines':relation(Machine, order_by=machines.c.machine_id)
+ })
+ mapper(Manager, managers,
+ inherits=Person, polymorphic_identity='manager')
+ mapper(Boss, boss, inherits=Manager, polymorphic_identity='boss')
+ mapper(Paperwork, paperwork)
+
+ @testing.resolve_artifact_names
+ def test_single_prop(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Company).join(Company.employees),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN people ON companies.company_id = people.company_id"
+ )
+
+ @testing.resolve_artifact_names
+ def test_single_prop_of_type(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Company).join(Company.employees.of_type(Engineer)),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN (SELECT people.person_id AS people_person_id, people.company_id AS "
+ "people_company_id, people.name AS people_name, people.type AS people_type, engineers.person_id AS "
+ "engineers_person_id, engineers.status AS engineers_status, engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language "
+ "FROM people JOIN engineers ON people.person_id = engineers.person_id) AS anon_1 ON companies.company_id = anon_1.people_company_id"
+ )
+
+ @testing.resolve_artifact_names
+ def test_prop_with_polymorphic(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Person).with_polymorphic(Manager).join('paperwork').filter(Paperwork.description.like('%review%')),
+ "SELECT people.person_id AS people_person_id, people.company_id AS people_company_id, "
+ "people.name AS people_name, people.type AS people_type, managers.person_id AS managers_person_id, "
+ "managers.status AS managers_status, managers.manager_name AS managers_manager_name FROM people "
+ "LEFT OUTER JOIN managers ON people.person_id = managers.person_id JOIN paperwork ON people.person_id = "
+ "paperwork.person_id WHERE paperwork.description LIKE :description_1 ORDER BY people.person_id"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(Person).with_polymorphic(Manager).join('paperwork', aliased=True).filter(Paperwork.description.like('%review%')),
+ "SELECT people.person_id AS people_person_id, people.company_id AS people_company_id, "
+ "people.name AS people_name, people.type AS people_type, managers.person_id AS managers_person_id, "
+ "managers.status AS managers_status, managers.manager_name AS managers_manager_name "
+ "FROM people LEFT OUTER JOIN managers ON people.person_id = managers.person_id JOIN "
+ "paperwork AS paperwork_1 ON people.person_id = paperwork_1.person_id "
+ "WHERE paperwork_1.description LIKE :description_1 ORDER BY people.person_id"
+ , use_default_dialect=True
+ )
+
+ @testing.resolve_artifact_names
+ def test_explicit_polymorphic_join(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Company).join(Engineer).filter(Engineer.engineer_name=='vlad'),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN (SELECT people.person_id AS people_person_id, people.company_id AS "
+ "people_company_id, people.name AS people_name, people.type AS people_type, engineers.person_id AS "
+ "engineers_person_id, engineers.status AS engineers_status, engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language "
+ "FROM people JOIN engineers ON people.person_id = engineers.person_id) AS anon_1 ON "
+ "companies.company_id = anon_1.people_company_id "
+ "WHERE anon_1.engineers_engineer_name = :engineer_name_1"
+ , use_default_dialect=True
+ )
+ self.assert_compile(
+ sess.query(Company).join((Engineer, Company.company_id==Engineer.company_id)).filter(Engineer.engineer_name=='vlad'),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN (SELECT people.person_id AS people_person_id, people.company_id AS "
+ "people_company_id, people.name AS people_name, people.type AS people_type, engineers.person_id AS "
+ "engineers_person_id, engineers.status AS engineers_status, engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language "
+ "FROM people JOIN engineers ON people.person_id = engineers.person_id) AS anon_1 ON "
+ "companies.company_id = anon_1.people_company_id "
+ "WHERE anon_1.engineers_engineer_name = :engineer_name_1"
+ , use_default_dialect=True
+ )
+
+ @testing.resolve_artifact_names
+ def test_multiple_adaption(self):
+ """test that multiple filter() adapters get chained together and work correctly within a multiple-entry join()."""
+
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Company).join((people.join(engineers), Company.employees)).
+ filter(Engineer.name=='dilbert'),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN (SELECT people.person_id AS people_person_id, people.company_id AS "
+ "people_company_id, people.name AS people_name, people.type AS people_type, engineers.person_id "
+ "AS engineers_person_id, engineers.status AS engineers_status, engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language FROM people JOIN engineers ON people.person_id = "
+ "engineers.person_id) AS anon_1 ON companies.company_id = anon_1.people_company_id WHERE anon_1.people_name = :name_1"
+ , use_default_dialect = True
+ )
+
+ mach_alias = machines.select()
+ self.assert_compile(
+ sess.query(Company).join((people.join(engineers), Company.employees), (mach_alias, Engineer.machines)).
+ filter(Engineer.name=='dilbert').filter(Machine.name=='foo'),
+ "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
+ "FROM companies JOIN (SELECT people.person_id AS people_person_id, people.company_id AS "
+ "people_company_id, people.name AS people_name, people.type AS people_type, engineers.person_id "
+ "AS engineers_person_id, engineers.status AS engineers_status, engineers.engineer_name AS engineers_engineer_name, "
+ "engineers.primary_language AS engineers_primary_language FROM people JOIN engineers ON people.person_id = "
+ "engineers.person_id) AS anon_1 ON companies.company_id = anon_1.people_company_id JOIN "
+ "(SELECT machines.machine_id AS machine_id, machines.name AS name, machines.engineer_id AS engineer_id "
+ "FROM machines) AS anon_2 ON anon_1.engineers_person_id = anon_2.engineer_id "
+ "WHERE anon_1.people_name = :name_1 AND anon_2.name = :name_2"
+ , use_default_dialect = True
+ )
+
+
+
+class JoinTest(QueryTest, AssertsCompiledSQL):
+
+ def test_foo(self):
+ sess = create_session()
+
+ def test_single_name(self):
+ sess = create_session()
-class JoinTest(QueryTest):
+ self.assert_compile(
+ sess.query(User).join("orders"),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users JOIN orders ON users.id = orders.user_id"
+ )
+
+ assert_raises(
+ sa_exc.InvalidRequestError,
+ sess.query(User).join, "user",
+ )
+
+ self.assert_compile(
+ sess.query(User).join("orders", "items"),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
+ "JOIN orders ON users.id = orders.user_id JOIN order_items AS order_items_1 "
+ "ON orders.id = order_items_1.order_id JOIN items ON items.id = order_items_1.item_id"
+ , use_default_dialect=True
+ )
+
+ # test overlapping paths. User->orders is used by both joins, but rendered once.
+ self.assert_compile(
+ sess.query(User).join("orders", "items").join("orders", "address"),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users JOIN orders "
+ "ON users.id = orders.user_id JOIN order_items AS order_items_1 ON orders.id = "
+ "order_items_1.order_id JOIN items ON items.id = order_items_1.item_id JOIN addresses "
+ "ON addresses.id = orders.address_id"
+ , use_default_dialect=True
+ )
+
+ def test_single_prop(self):
+ sess = create_session()
+ self.assert_compile(
+ sess.query(User).join(User.orders),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users JOIN orders ON users.id = orders.user_id"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User).join(Order.user),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM orders JOIN users ON users.id = orders.user_id"
+ , use_default_dialect=True
+ )
+
+ oalias1 = aliased(Order)
+ oalias2 = aliased(Order)
+ self.assert_compile(
+ sess.query(User).join(oalias1.user),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM orders AS orders_1 JOIN users ON users.id = orders_1.user_id"
+ , use_default_dialect=True
+ )
+
+ # another nonsensical query. (from [ticket:1537]).
+ # in this case, the contract of "left to right" is honored
+ self.assert_compile(
+ sess.query(User).join(oalias1.user).join(oalias2.user),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM orders AS orders_1 JOIN users ON users.id = orders_1.user_id, "
+ "orders AS orders_2 JOIN users ON users.id = orders_2.user_id"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User).join(User.orders, Order.items),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
+ "JOIN orders ON users.id = orders.user_id JOIN order_items AS order_items_1 "
+ "ON orders.id = order_items_1.order_id JOIN items ON items.id = order_items_1.item_id"
+ , use_default_dialect=True
+ )
+
+ ualias = aliased(User)
+ self.assert_compile(
+ sess.query(ualias).join(ualias.orders),
+ "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name "
+ "FROM users AS users_1 JOIN orders ON users_1.id = orders.user_id"
+ , use_default_dialect=True
+ )
+
+ # this query is somewhat nonsensical. the old system didn't render a correct
+ # query for this. In this case its the most faithful to what was asked -
+ # there's no linkage between User.orders and "oalias", so two FROM elements
+ # are generated.
+ oalias = aliased(Order)
+ self.assert_compile(
+ sess.query(User).join(User.orders, oalias.items),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
+ "JOIN orders ON users.id = orders.user_id, "
+ "orders AS orders_1 JOIN order_items AS order_items_1 ON orders_1.id = order_items_1.order_id "
+ "JOIN items ON items.id = order_items_1.item_id"
+ , use_default_dialect=True
+ )
+
+ # same as before using an aliased() for User as well
+ ualias = aliased(User)
+ self.assert_compile(
+ sess.query(ualias).join(ualias.orders, oalias.items),
+ "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name FROM users AS users_1 "
+ "JOIN orders ON users_1.id = orders.user_id, "
+ "orders AS orders_1 JOIN order_items AS order_items_1 ON orders_1.id = order_items_1.order_id "
+ "JOIN items ON items.id = order_items_1.item_id"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User).filter(User.name=='ed').from_self().join(User.orders),
+ "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS anon_1_users_name "
+ "FROM (SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users "
+ "WHERE users.name = :name_1) AS anon_1 JOIN orders ON anon_1.users_id = orders.user_id"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User).join(User.addresses, aliased=True).filter(Address.email_address=='foo'),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users JOIN addresses AS addresses_1 ON users.id = addresses_1.user_id "
+ "WHERE addresses_1.email_address = :email_address_1"
+ , use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User).join(User.orders, Order.items, aliased=True).filter(Item.id==10),
+ "SELECT users.id AS users_id, users.name AS users_name "
+ "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
+ "JOIN order_items AS order_items_1 ON orders_1.id = order_items_1.order_id "
+ "JOIN items AS items_1 ON items_1.id = order_items_1.item_id "
+ "WHERE items_1.id = :id_1"
+ , use_default_dialect=True
+ )
+
def test_overlapping_paths(self):
for aliased in (True,False):
# load a user who has an order that contains item id 3 and address id 1 (order 3, owned by jack)
- result = create_session().query(User).join(['orders', 'items'], aliased=aliased).filter_by(id=3).join(['orders','address'], aliased=aliased).filter_by(id=1).all()
+ result = create_session().query(User).join(['orders', 'items'], aliased=aliased).\
+ filter_by(id=3).join(['orders','address'], aliased=aliased).filter_by(id=1).all()
assert [User(id=7, name='jack')] == result
def test_overlapping_paths_outerjoin(self):
- result = create_session().query(User).outerjoin(['orders', 'items']).filter_by(id=3).outerjoin(['orders','address']).filter_by(id=1).all()
+ result = create_session().query(User).outerjoin(['orders', 'items']).\
+ filter_by(id=3).outerjoin(['orders','address']).filter_by(id=1).all()
assert [User(id=7, name='jack')] == result
def test_from_joinpoint(self):
ualias = aliased(User)
oalias1 = aliased(Order)
oalias2 = aliased(Order)
- result = sess.query(ualias).join((oalias1, ualias.orders), (oalias2, ualias.orders)).\
- filter(or_(oalias1.user_id==9, oalias2.user_id==7)).all()
- eq_(result, [User(id=7,name=u'jack'), User(id=9,name=u'fred')])
+ self.assert_compile(
+ sess.query(ualias).join((oalias1, ualias.orders), (oalias2, ualias.orders)).\
+ filter(or_(oalias1.user_id==9, oalias2.user_id==7)),
+ "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name FROM users AS users_1 "
+ "JOIN orders AS orders_1 ON users_1.id = orders_1.user_id JOIN orders AS orders_2 ON "
+ "users_1.id = orders_2.user_id WHERE orders_1.user_id = :user_id_1 OR orders_2.user_id = :user_id_2",
+ use_default_dialect=True
+ )
+
+ def test_select_from_orm_joins(self):
+ sess = create_session()
+
+ ualias = aliased(User)
+ oalias1 = aliased(Order)
+ oalias2 = aliased(Order)
+
+ self.assert_compile(
+ join(User, oalias2, User.id==oalias2.user_id),
+ "users JOIN orders AS orders_1 ON users.id = orders_1.user_id",
+ use_default_dialect=True
+ )
+
+ self.assert_compile(
+ join(ualias, oalias1, ualias.orders),
+ "users AS users_1 JOIN orders AS orders_1 ON users_1.id = orders_1.user_id",
+ use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(ualias).select_from(join(ualias, oalias1, ualias.orders)),
+ "SELECT users_1.id AS users_1_id, users_1.name AS users_1_name FROM users AS users_1 "
+ "JOIN orders AS orders_1 ON users_1.id = orders_1.user_id",
+ use_default_dialect=True
+ )
+
+ self.assert_compile(
+ sess.query(User, ualias).select_from(join(ualias, oalias1, ualias.orders)),
+ "SELECT users.id AS users_id, users.name AS users_name, users_1.id AS users_1_id, "
+ "users_1.name AS users_1_name FROM users, users AS users_1 JOIN orders AS orders_1 ON users_1.id = orders_1.user_id",
+ use_default_dialect=True
+ )
+
+ # this fails (and we cant quite fix right now).
+ if False:
+ self.assert_compile(
+ sess.query(User, ualias).\
+ join((oalias1, ualias.orders)).\
+ join((oalias2, User.id==oalias2.user_id)).\
+ filter(or_(oalias1.user_id==9, oalias2.user_id==7)),
+ "SELECT users.id AS users_id, users.name AS users_name, users_1.id AS users_1_id, users_1.name AS "
+ "users_1_name FROM users JOIN orders AS orders_2 ON users.id = orders_2.user_id, "
+ "users AS users_1 JOIN orders AS orders_1 ON users_1.id = orders_1.user_id "
+ "WHERE orders_1.user_id = :user_id_1 OR orders_2.user_id = :user_id_2",
+ use_default_dialect=True
+ )
+
+ # this is the same thing using explicit orm.join() (which now offers multiple again)
+ self.assert_compile(
+ sess.query(User, ualias).\
+ select_from(
+ join(ualias, oalias1, ualias.orders),
+ join(User, oalias2, User.id==oalias2.user_id),
+ ).\
+ filter(or_(oalias1.user_id==9, oalias2.user_id==7)),
+ "SELECT users.id AS users_id, users.name AS users_name, users_1.id AS users_1_id, users_1.name AS "
+ "users_1_name FROM users AS users_1 JOIN orders AS orders_1 ON users_1.id = orders_1.user_id, "
+ "users JOIN orders AS orders_2 ON users.id = orders_2.user_id "
+ "WHERE orders_1.user_id = :user_id_1 OR orders_2.user_id = :user_id_2",
+
+ use_default_dialect=True
+ )
+
+
+ def test_overlapping_backwards_joins(self):
+ sess = create_session()
+
+ oalias1 = aliased(Order)
+ oalias2 = aliased(Order)
+
+ # this is invalid SQL - joins from orders_1/orders_2 to User twice.
+ # but that is what was asked for so they get it !
+ self.assert_compile(
+ sess.query(User).join(oalias1.user).join(oalias2.user),
+ "SELECT users.id AS users_id, users.name AS users_name FROM orders AS orders_1 "
+ "JOIN users ON users.id = orders_1.user_id, orders AS orders_2 JOIN users ON users.id = orders_2.user_id",
+ use_default_dialect=True,
+ )
+
+ def test_replace_multiple_from_clause(self):
+ """test adding joins onto multiple FROM clauses"""
+
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(Address, User).join(Address.dingaling).join(User.orders, Order.items),
+ "SELECT addresses.id AS addresses_id, addresses.user_id AS addresses_user_id, "
+ "addresses.email_address AS addresses_email_address, users.id AS users_id, "
+ "users.name AS users_name FROM addresses JOIN dingalings ON addresses.id = dingalings.address_id, "
+ "users JOIN orders ON users.id = orders.user_id JOIN order_items AS order_items_1 "
+ "ON orders.id = order_items_1.order_id JOIN items ON items.id = order_items_1.item_id",
+ use_default_dialect = True
+ )
+ def test_multiple_adaption(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(User).join(User.orders, Order.items, aliased=True).filter(Order.id==7).filter(Item.id==8),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users JOIN orders AS orders_1 "
+ "ON users.id = orders_1.user_id JOIN order_items AS order_items_1 ON orders_1.id = order_items_1.order_id "
+ "JOIN items AS items_1 ON items_1.id = order_items_1.item_id WHERE orders_1.id = :id_1 AND items_1.id = :id_2",
+ use_default_dialect=True
+ )
+
+ def test_onclause_conditional_adaption(self):
+ sess = create_session()
+
+ self.assert_compile(
+ sess.query(User).join(User.orders,
+ (Item,
+ and_(Order.id==order_items.c.order_id, order_items.c.item_id==Item.id)
+ ),aliased=True
+ ),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users JOIN "
+ "orders AS orders_1 ON users.id = orders_1.user_id JOIN items AS items_1 "
+ "ON orders_1.id = order_items.order_id AND order_items.item_id = items_1.id",
+ use_default_dialect=True
+ )
+
+ oalias = orders.select()
+ self.assert_compile(
+ sess.query(User).join((oalias, User.orders),
+ (Item,
+ and_(Order.id==order_items.c.order_id, order_items.c.item_id==Item.id)
+ ),
+ ),
+ "SELECT users.id AS users_id, users.name AS users_name FROM users JOIN "
+ "(SELECT orders.id AS id, orders.user_id AS user_id, orders.address_id AS address_id, orders.description "
+ "AS description, orders.isopen AS isopen FROM orders) AS anon_1 ON users.id = anon_1.user_id JOIN items "
+ "ON anon_1.id = order_items.order_id AND order_items.item_id = items.id",
+ use_default_dialect=True
+ )
+
+
+ # query.join(<stuff>, aliased=True).join((target, sql_expression))
+ # or: query.join(path_to_some_joined_table_mapper).join((target, sql_expression))
+
def test_pure_expression_error(self):
sess = create_session()
)
def test_plain_table(self):
-
+
sess = create_session()
eq_(
)
-class MultiplePathTest(_base.MappedTest):
+class MultiplePathTest(_base.MappedTest, AssertsCompiledSQL):
@classmethod
def define_tables(cls, metadata):
global t1, t2, t1t2_1, t1t2_2
})
mapper(T2, t2)
- q = create_session().query(T1).join('t2s_1').filter(t2.c.id==5).reset_joinpoint()
- assert_raises_message(sa_exc.InvalidRequestError, "a path to this table along a different secondary table already exists.",
- q.join, 't2s_2'
+ q = create_session().query(T1).join('t2s_1').filter(t2.c.id==5).reset_joinpoint().join('t2s_2')
+ self.assert_compile(
+ q,
+ "SELECT t1.id AS t1_id, t1.data AS t1_data FROM t1 JOIN t1t2_1 AS t1t2_1_1 "
+ "ON t1.id = t1t2_1_1.t1id JOIN t2 ON t2.id = t1t2_1_1.t2id JOIN t1t2_2 AS t1t2_2_1 "
+ "ON t1.id = t1t2_2_1.t1id JOIN t2 ON t2.id = t1t2_2_1.t2id WHERE t2.id = :id_1"
+ , use_default_dialect=True
)
- create_session().query(T1).join('t2s_1', aliased=True).filter(t2.c.id==5).reset_joinpoint().join('t2s_2').all()
- create_session().query(T1).join('t2s_1').filter(t2.c.id==5).reset_joinpoint().join('t2s_2', aliased=True).all()
class SynonymTest(QueryTest):