# determine the immediate parent class we are joining from,
# which needs to be aliased.
- if len(to_join) > 1:
- info = inspect(to_join[-1][0])
if len(to_join) < 2:
# in the case of a one level eager load, this is the
# leftmost "left_alias".
parent_alias = left_alias
- elif info.mapper.isa(self.parent):
- # In the case of multiple levels, retrieve
- # it from subq_path[-2]. This is the same as self.parent
- # in the vast majority of cases, and [ticket:2014]
- # illustrates a case where sub_path[-2] is a subclass
- # of self.parent
- parent_alias = orm_util.AliasedClass(
- to_join[-1][0],
- use_mapper_path=True)
else:
- # if of_type() were used leading to this relationship,
- # self.parent is more specific than subq_path[-2]
- parent_alias = orm_util.AliasedClass(
- self.parent,
- use_mapper_path=True)
+ info = inspect(to_join[-1][0])
+ if info.is_aliased_class:
+ parent_alias = info.entity
+ else:
+ # alias a plain mapper as we may be
+ # joining multiple times
+ parent_alias = orm_util.AliasedClass(
+ info.entity,
+ use_mapper_path=True)
local_cols = self.parent_property.local_columns
def _apply_joins(
self, q, to_join, left_alias, parent_alias,
effective_entity):
- for i, (mapper, key) in enumerate(to_join):
-
- # we need to use query.join() as opposed to
- # orm.join() here because of the
- # rich behavior it brings when dealing with
- # "with_polymorphic" mappers. "aliased"
- # and "from_joinpoint" take care of most of
- # the chaining and aliasing for us.
-
- first = i == 0
- middle = i < len(to_join) - 1
- second_to_last = i == len(to_join) - 2
- last = i == len(to_join) - 1
-
- if first:
- attr = getattr(left_alias, key)
- if last and effective_entity is not self.mapper:
- attr = attr.of_type(effective_entity)
- else:
- if last:
- attr = getattr(parent_alias, key).\
- of_type(effective_entity)
+
+ ltj = len(to_join)
+ if ltj == 1:
+ to_join = [
+ getattr(left_alias, to_join[0][1]).of_type(effective_entity)
+ ]
+ elif ltj == 2:
+ to_join = [
+ getattr(left_alias, to_join[0][1]).of_type(parent_alias),
+ getattr(parent_alias, to_join[-1][1]).of_type(effective_entity)
+ ]
+ elif ltj > 2:
+ middle = [
+ (
+ orm_util.AliasedClass(item[0])
+ if not inspect(item[0]).is_aliased_class
+ else item[0].entity,
+ item[1]
+ ) for item in to_join[1:-1]
+ ]
+ inner = []
+
+ while middle:
+ item = middle.pop(0)
+ attr = getattr(item[0], item[1])
+ if middle:
+ attr = attr.of_type(middle[0][0])
else:
- attr = getattr(mapper.entity, key)
+ attr = attr.of_type(parent_alias)
- if second_to_last:
- q = q.join(parent_alias, attr, from_joinpoint=True)
- else:
- q = q.join(attr, aliased=middle, from_joinpoint=True)
+ inner.append(attr)
+
+ to_join = [
+ getattr(left_alias, to_join[0][1]).of_type(inner[0].parent)
+ ] + inner + [
+ getattr(parent_alias, to_join[-1][1]).of_type(effective_entity)
+ ]
+
+ for attr in to_join:
+ q = q.join(attr, from_joinpoint=True)
return q
def _setup_options(self, q, subq_path, orig_query, effective_entity):
Machine, Paperwork, _PolymorphicFixtureBase, _Polymorphic,\
_PolymorphicPolymorphic, _PolymorphicUnions, _PolymorphicJoins,\
_PolymorphicAliasedJoins
+from sqlalchemy.testing.assertsql import AllOf, CompiledSQL
class _PolymorphicTestBase(object):
"AS anon_1 ON data_container.id = anon_1.job_container_id"
)
+
+class SubclassRelationshipTest2(
+ testing.AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
+
+ run_setup_classes = 'once'
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
+ __dialect__ = 'default'
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(Base):
+ __tablename__ = 't_a'
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
+ class B(Base):
+ __tablename__ = 't_b'
+
+ type = Column(String(2))
+ __mapper_args__ = {
+ 'polymorphic_identity': 'b',
+ 'polymorphic_on': type
+ }
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
+ # Relationship to A
+ a_id = Column(Integer, ForeignKey('t_a.id'))
+ a = relationship('A', backref='bs')
+
+ class B2(B):
+ __tablename__ = 't_b2'
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'b2',
+ }
+
+ id = Column(Integer, ForeignKey('t_b.id'), primary_key=True)
+
+ class C(Base):
+ __tablename__ = 't_c'
+
+ type = Column(String(2))
+ __mapper_args__ = {
+ 'polymorphic_identity': 'c',
+ 'polymorphic_on': type
+ }
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
+ # Relationship to B
+ b_id = Column(Integer, ForeignKey('t_b.id'))
+ b = relationship('B', backref='cs')
+
+ class C2(C):
+ __tablename__ = 't_c2'
+
+ __mapper_args__ = {
+ 'polymorphic_identity': 'c2',
+ }
+
+ id = Column(Integer, ForeignKey('t_c.id'), primary_key=True)
+
+ class D(Base):
+ __tablename__ = 't_d'
+
+ id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+
+ # Relationship to B
+ c_id = Column(Integer, ForeignKey('t_c.id'))
+ c = relationship('C', backref='ds')
+
+ @classmethod
+ def insert_data(cls):
+ s = Session(testing.db)
+
+ s.add_all(cls._fixture())
+ s.commit()
+
+ @classmethod
+ def _fixture(cls):
+ A, B, B2, C, C2, D = cls.classes('A', 'B', 'B2', 'C', 'C2', 'D')
+
+ return [
+ A(bs=[B2(cs=[C2(ds=[D()])])]),
+ A(bs=[B2(cs=[C2(ds=[D()])])]),
+ ]
+
+ def test_all_subq_query(self):
+ A, B, B2, C, C2, D = self.classes('A', 'B', 'B2', 'C', 'C2', 'D')
+
+ session = Session(testing.db)
+
+ b_b2 = with_polymorphic(B, [B2], flat=True)
+ c_c2 = with_polymorphic(C, [C2], flat=True)
+
+ q = session.query(
+ A
+ ).options(
+ subqueryload(
+ A.bs.of_type(b_b2)
+ ).subqueryload(
+ b_b2.cs.of_type(c_c2)
+ ).subqueryload(
+ c_c2.ds
+ )
+ )
+
+ self.assert_sql_execution(
+ testing.db,
+ q.all,
+ CompiledSQL(
+ "SELECT t_a.id AS t_a_id FROM t_a",
+ {}
+ ),
+ CompiledSQL(
+ "SELECT t_b_1.type AS t_b_1_type, t_b_1.id AS t_b_1_id, "
+ "t_b_1.a_id AS t_b_1_a_id, t_b2_1.id AS t_b2_1_id, "
+ "anon_1.t_a_id AS anon_1_t_a_id FROM "
+ "(SELECT t_a.id AS t_a_id FROM t_a) AS anon_1 "
+ "JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+ "ON t_b_1.id = t_b2_1.id) ON anon_1.t_a_id = t_b_1.a_id "
+ "ORDER BY anon_1.t_a_id",
+ {}
+ ),
+ CompiledSQL(
+ "SELECT t_c_1.type AS t_c_1_type, t_c_1.id AS t_c_1_id, "
+ "t_c_1.b_id AS t_c_1_b_id, t_c2_1.id AS t_c2_1_id, "
+ "t_b_1.id AS t_b_1_id FROM (SELECT t_a.id AS t_a_id FROM t_a) "
+ "AS anon_1 JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+ "ON t_b_1.id = t_b2_1.id) ON anon_1.t_a_id = t_b_1.a_id "
+ "JOIN (t_c AS t_c_1 LEFT OUTER JOIN t_c2 AS t_c2_1 ON "
+ "t_c_1.id = t_c2_1.id) ON t_b_1.id = t_c_1.b_id "
+ "ORDER BY t_b_1.id",
+ {}
+ ),
+ CompiledSQL(
+ "SELECT t_d.id AS t_d_id, t_d.c_id AS t_d_c_id, "
+ "t_c_1.id AS t_c_1_id "
+ "FROM (SELECT t_a.id AS t_a_id FROM t_a) AS anon_1 "
+ "JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+ "ON t_b_1.id = t_b2_1.id) "
+ "ON anon_1.t_a_id = t_b_1.a_id "
+ "JOIN (t_c AS t_c_1 LEFT OUTER JOIN t_c2 AS t_c2_1 "
+ "ON t_c_1.id = t_c2_1.id) "
+ "ON t_b_1.id = t_c_1.b_id "
+ "JOIN t_d ON t_c_1.id = t_d.c_id ORDER BY t_c_1.id",
+ {}
+ )
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
