# This is basically the only place that the "is_unnatural" flag
# actually changes behavior.
if parent.path and (self.is_aliased_class or parent.is_unnatural):
- # this is an infrequent code path used only for loader strategies
- # that also make use of of_type().
- if entity.mapper.isa(parent.natural_path[-1].mapper): # type: ignore # noqa: E501
+ # this is an infrequent code path used for loader strategies that
+ # also make use of of_type() or other intricate polymorphic
+ # base/subclass combinations
+ parent_natural_entity = parent.natural_path[-1]
+
+ if entity.mapper.isa(
+ parent_natural_entity.mapper # type: ignore
+ ) or parent_natural_entity.mapper.isa( # type: ignore
+ entity.mapper
+ ):
+ # when the entity mapper and parent mapper are in an
+ # inheritance relationship, use entity.mapper in natural_path.
+ # First case: entity.mapper inherits from parent mapper (e.g.,
+ # accessing a subclass mapper through parent path). Second case
+ # (issue #13193): parent mapper inherits from entity.mapper
+ # (e.g., parent path has Sub(Base) but we're accessing with
+ # Base where Base.related is declared, so use Base in
+ # natural_path).
self.natural_path = parent.natural_path + (entity.mapper,)
else:
self.natural_path = parent.natural_path + (
- parent.natural_path[-1].entity, # type: ignore
+ parent_natural_entity.entity, # type: ignore
)
# it seems to make sense that since these paths get mixed up
# with statements that are cached or not, we should make
self.assert_sql_count(testing.db, go, 2)
+ def test_chained_joinedload_with_polymorphic_inherited_rel(self):
+ """test #13193
+
+ Chained joinedload where the final relationship is declared on a
+ base mapper but accessed through a subclass in a with_polymorphic
+ query. The loader should correctly find the option even though the
+ lookup path uses the declaring mapper.
+ """
+ sess = fixture_session()
+
+ def go():
+ # paperwork is declared on Person but we're accessing via Engineer
+ # Using with_polymorphic for Company->employees and chaining to
+ # Engineer.paperwork which is inherited from Person.paperwork
+ result = (
+ sess.query(Company)
+ .options(
+ joinedload(Company.employees.of_type(Engineer)).joinedload(
+ Engineer.paperwork
+ )
+ )
+ .filter(Company.company_id == 1)
+ .all()
+ )
+ eq_(len(result), 1)
+ company = result[0]
+ engineers = [
+ e for e in company.employees if isinstance(e, Engineer)
+ ]
+ # Verify engineers loaded (should be > 0)
+ assert len(engineers) > 0
+ # Access paperwork - should not trigger additional SQL
+ for eng in engineers:
+ _ = list(eng.paperwork)
+
+ # Should be 1 query with all joins including paperwork
+ self.assert_sql_count(testing.db, go, 1)
+
def test_query_subclass_join_to_base_relationship(self):
sess = fixture_session()
# non-polymorphic
],
)
)
+
+
+class ChainedJoinedloadInheritedRelationshipTest(
+ fixtures.MappedTest, AssertsCompiledSQL
+):
+ """test #13193 - chained joinedload with inherited relationships
+
+ When using with_polymorphic() and chained joinedload() where the final
+ relationship is inherited from a base mapper, verify that the loader
+ is correctly found and applied.
+ """
+
+ __dialect__ = "default"
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "meta",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("name", String(50)),
+ )
+
+ Table(
+ "top",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("type", String(50)),
+ Column("meta_id", Integer, ForeignKey("meta.id")),
+ )
+
+ Table(
+ "foo",
+ metadata,
+ Column("id", Integer, ForeignKey("top.id"), primary_key=True),
+ )
+
+ Table(
+ "bar",
+ metadata,
+ Column("id", Integer, ForeignKey("top.id"), primary_key=True),
+ Column("foo_id", Integer, ForeignKey("foo.id")),
+ )
+
+ @classmethod
+ def setup_classes(cls):
+ class Meta(cls.Comparable):
+ pass
+
+ class Top(cls.Comparable):
+ pass
+
+ class Foo(Top):
+ pass
+
+ class Bar(Top):
+ pass
+
+ @classmethod
+ def setup_mappers(cls):
+ Meta, Top, Foo, Bar = cls.classes("Meta", "Top", "Foo", "Bar")
+ meta, top, foo, bar = cls.tables("meta", "top", "foo", "bar")
+
+ cls.mapper_registry.map_imperatively(Meta, meta)
+
+ cls.mapper_registry.map_imperatively(
+ Top,
+ top,
+ polymorphic_on=top.c.type,
+ properties={"meta": relationship(Meta)},
+ )
+
+ cls.mapper_registry.map_imperatively(
+ Foo,
+ foo,
+ inherits=Top,
+ polymorphic_identity="FOO",
+ )
+
+ cls.mapper_registry.map_imperatively(
+ Bar,
+ bar,
+ inherits=Top,
+ polymorphic_identity="BAR",
+ properties={"foo": relationship(Foo, foreign_keys=[bar.c.foo_id])},
+ )
+
+ @classmethod
+ def insert_data(cls, connection):
+ meta, top, foo, bar = cls.tables("meta", "top", "foo", "bar")
+
+ connection.execute(meta.insert(), [{"id": 1, "name": "meta1"}])
+ connection.execute(
+ top.insert(),
+ [
+ {"id": 1, "type": "FOO", "meta_id": 1},
+ {"id": 2, "type": "BAR", "meta_id": 1},
+ ],
+ )
+ connection.execute(foo.insert(), [{"id": 1}])
+ connection.execute(bar.insert(), [{"id": 2, "foo_id": 1}])
+
+ @testing.variation("use_of_type", ["plain", "aliased"])
+ def test_chained_joinedload_inherited_relationship(self, use_of_type):
+ """Test chained joinedload where final rel is inherited from base.
+
+ The 'meta' relationship is declared on Top but accessed via Foo.
+ The loader should be found whether we use plain Foo or aliased(Foo).
+ """
+ Top, Foo, Bar = self.classes("Top", "Foo", "Bar")
+ sess = fixture_session()
+
+ tp = with_polymorphic(Top, "*", flat=True)
+
+ if use_of_type.plain:
+ # joinedload(tp.Bar.foo).joinedload(Foo.meta)
+ options = [joinedload(tp.Bar.foo).joinedload(Foo.meta)]
+ elif use_of_type.aliased:
+ # joinedload(tp.Bar.foo.of_type(foo_alias)).joinedload(foo_alias.meta)
+ foo_alias = aliased(Foo, flat=True)
+ options = [
+ joinedload(tp.Bar.foo.of_type(foo_alias)).joinedload(
+ foo_alias.meta
+ )
+ ]
+
+ # Verify SQL includes join to meta table
+ with self.sql_execution_asserter(testing.db) as asserter:
+ result = sess.query(tp).options(*options).all()
+
+ # Verify round-trip: data loads correctly
+ eq_(len(result), 2)
+ bar_obj = [r for r in result if isinstance(r, Bar)][0]
+ eq_(bar_obj.foo.meta.name, "meta1")
+
+ # Should have exactly one query with the meta join
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT top_1.id AS top_1_id, top_1.type AS top_1_type, "
+ "top_1.meta_id AS top_1_meta_id, "
+ "foo_1.id AS foo_1_id, bar_1.id AS bar_1_id, "
+ "bar_1.foo_id AS bar_1_foo_id, "
+ "meta_1.id AS meta_1_id, meta_1.name AS meta_1_name, "
+ "foo_2.id AS foo_2_id, top_2.id AS top_2_id, "
+ "top_2.type AS top_2_type, top_2.meta_id AS top_2_meta_id "
+ "FROM top AS top_1 "
+ "LEFT OUTER JOIN foo AS foo_1 ON top_1.id = foo_1.id "
+ "LEFT OUTER JOIN bar AS bar_1 ON top_1.id = bar_1.id "
+ "LEFT OUTER JOIN "
+ "(top AS top_2 JOIN foo AS foo_2 ON top_2.id = foo_2.id) "
+ "ON foo_2.id = bar_1.foo_id "
+ "LEFT OUTER JOIN meta AS meta_1 "
+ "ON meta_1.id = top_2.meta_id"
+ )
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
