class PropRegistry(PathRegistry):
+ is_unnatural = False
+
def __init__(self, parent, prop):
# restate this path in terms of the
# given MapperProperty's parent.
insp = inspection.inspect(parent[-1])
+ natural_parent = parent
if not insp.is_aliased_class or insp._use_mapper_path:
- parent = parent.parent[prop.parent]
+ parent = natural_parent = parent.parent[prop.parent]
elif insp.is_aliased_class and insp.with_polymorphic_mappers:
if (
prop.parent is not insp.mapper
and prop.parent in insp.with_polymorphic_mappers
):
subclass_entity = parent[-1]._entity_for_mapper(prop.parent)
- if subclass_entity._use_mapper_path:
- parent = parent.parent[subclass_entity.mapper]
+ parent = parent.parent[subclass_entity]
+
+ # when building a path where with_polymorphic() is in use,
+ # special logic to determine the "natural path" when subclass
+ # entities are used.
+ #
+ # here we are trying to distinguish between a path that starts
+ # on a the with_polymorhpic entity vs. one that starts on a
+ # normal entity that introduces a with_polymorphic() in the
+ # middle using of_type():
+ #
+ # # as in test_polymorphic_rel->
+ # # test_subqueryload_on_subclass_uses_path_correctly
+ # wp = with_polymorphic(RegularEntity, "*")
+ # sess.query(wp).options(someload(wp.SomeSubEntity.foos))
+ #
+ # vs
+ #
+ # # as in test_relationship->JoinedloadWPolyOfTypeContinued
+ # wp = with_polymorphic(SomeFoo, "*")
+ # sess.query(RegularEntity).options(
+ # someload(RegularEntity.foos.of_type(wp))
+ # .someload(wp.SubFoo.bar)
+ # )
+ #
+ # in the former case, the Query as it generates a path that we
+ # want to match will be in terms of the with_polymorphic at the
+ # beginning. in the latter case, Query will generate simple
+ # paths that don't know about this with_polymorphic, so we must
+ # use a separate natural path.
+ #
+ #
+ if parent.parent:
+ natural_parent = parent.parent[subclass_entity.mapper]
+ self.is_unnatural = True
else:
- parent = parent.parent[subclass_entity]
+ natural_parent = parent
self.prop = prop
self.parent = parent
self.path = parent.path + (prop,)
- self.natural_path = parent.natural_path + (prop,)
+ self.natural_path = natural_parent.natural_path + (prop,)
self._wildcard_path_loader_key = (
"loader",
# "enhanced" path in self.path and the "natural" path that doesn't
# include those objects so these two traversals can be matched up.
- if parent.path and self.is_aliased_class:
+ # the test here for "(self.is_aliased_class or parent.is_unnatural)"
+ # are to avoid the more expensive conditional logic that follows if we
+ # know we don't have to do it. This conditional can just as well be
+ # "if parent.path:", it just is more function calls.
+ 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].entity):
selectable,
base_alias=self,
adapt_on_names=adapt_on_names,
- use_mapper_path=True,
+ use_mapper_path=_use_mapper_path,
)
setattr(self.entity, poly.class_.__name__, ent)
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
+from sqlalchemy import util
from sqlalchemy.orm import create_session
from sqlalchemy.orm import mapper
from sqlalchemy.orm import polymorphic_union
mapper(Machine, machines)
- person_with_polymorphic, manager_with_polymorphic = (
- cls._get_polymorphics()
- )
+ (
+ person_with_polymorphic,
+ manager_with_polymorphic,
+ ) = cls._get_polymorphics()
mapper(
Person,
cls.tables.boss,
)
person_join = polymorphic_union(
- {
- "engineer": people.join(engineers),
- "manager": people.join(managers),
- },
+ util.OrderedDict(
+ [
+ ("engineer", people.join(engineers)),
+ ("manager", people.join(managers)),
+ ]
+ ),
None,
"pjoin",
)
from sqlalchemy.orm import with_polymorphic
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import eq_
+from sqlalchemy.testing.assertsql import CompiledSQL
from ._poly_fixtures import _Polymorphic
from ._poly_fixtures import _PolymorphicAliasedJoins
from ._poly_fixtures import _PolymorphicJoins
]
def go():
- # test load People with subqueryload to engineers + machines
eq_(
sess.query(Person)
.with_polymorphic("*")
class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
- pass
+ def test_subqueryload_on_subclass_uses_path_correctly(self):
+ sess = create_session()
+ expected = [
+ Engineer(
+ name="dilbert",
+ engineer_name="dilbert",
+ primary_language="java",
+ status="regular engineer",
+ machines=[
+ Machine(name="IBM ThinkPad"),
+ Machine(name="IPhone"),
+ ],
+ )
+ ]
+
+ with self.sql_execution_asserter(testing.db) as asserter:
+ wp = with_polymorphic(Person, "*")
+ eq_(
+ sess.query(wp)
+ .options(subqueryload(wp.Engineer.machines))
+ .filter(wp.name == "dilbert")
+ .all(),
+ expected,
+ )
+
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT pjoin.person_id AS pjoin_person_id, "
+ "pjoin.company_id AS pjoin_company_id, "
+ "pjoin.name AS pjoin_name, pjoin.type AS pjoin_type, "
+ "pjoin.status AS pjoin_status, "
+ "pjoin.engineer_name AS pjoin_engineer_name, "
+ "pjoin.primary_language AS pjoin_primary_language, "
+ "pjoin.manager_name AS pjoin_manager_name "
+ "FROM (SELECT engineers.person_id AS person_id, "
+ "people.company_id AS company_id, people.name AS name, "
+ "people.type AS type, engineers.status AS status, "
+ "engineers.engineer_name AS engineer_name, "
+ "engineers.primary_language AS primary_language, "
+ "CAST(NULL AS VARCHAR(50)) AS manager_name "
+ "FROM people JOIN engineers ON people.person_id = "
+ "engineers.person_id UNION ALL SELECT managers.person_id "
+ "AS person_id, people.company_id AS company_id, people.name "
+ "AS name, people.type AS type, managers.status AS status, "
+ "CAST(NULL AS VARCHAR(50)) AS engineer_name, "
+ "CAST(NULL AS VARCHAR(50)) AS primary_language, "
+ "managers.manager_name AS manager_name FROM people "
+ "JOIN managers ON people.person_id = managers.person_id) "
+ "AS pjoin WHERE pjoin.name = :name_1",
+ params=[{"name_1": "dilbert"}],
+ ),
+ CompiledSQL(
+ "SELECT machines.machine_id AS machines_machine_id, "
+ "machines.name AS machines_name, machines.engineer_id "
+ "AS machines_engineer_id, anon_1.pjoin_person_id AS "
+ "anon_1_pjoin_person_id FROM "
+ "(SELECT pjoin.person_id AS pjoin_person_id FROM "
+ "(SELECT engineers.person_id AS person_id, people.company_id "
+ "AS company_id, people.name AS name, "
+ "people.type AS type, engineers.status AS status, "
+ "engineers.engineer_name AS engineer_name, "
+ "engineers.primary_language AS primary_language, "
+ "CAST(NULL AS VARCHAR(50)) AS manager_name FROM people "
+ "JOIN engineers ON people.person_id = engineers.person_id "
+ "UNION ALL SELECT managers.person_id AS person_id, "
+ "people.company_id AS company_id, people.name AS name, "
+ "people.type AS type, managers.status AS status, "
+ "CAST(NULL AS VARCHAR(50)) AS engineer_name, "
+ "CAST(NULL AS VARCHAR(50)) AS primary_language, "
+ "managers.manager_name AS manager_name FROM people "
+ "JOIN managers ON people.person_id = managers.person_id) "
+ "AS pjoin WHERE pjoin.name = :name_1) AS anon_1 JOIN "
+ "machines ON anon_1.pjoin_person_id = machines.engineer_id "
+ "ORDER BY anon_1.pjoin_person_id, machines.machine_id",
+ params=[{"name_1": "dilbert"}],
+ ),
+ )
class PolymorphicAliasedJoinsTest(
)
def test_with_poly_sub(self):
+ Company = _poly_fixtures.Company
Person = _poly_fixtures.Person
Engineer = _poly_fixtures.Engineer
emapper = inspect(Engineer)
+ cmapper = inspect(Company)
p_poly = with_polymorphic(Person, [Engineer])
- e_poly = inspect(p_poly.Engineer) # noqa - used by comment below
+ e_poly_insp = inspect(p_poly.Engineer) # noqa - used by comment below
p_poly_insp = inspect(p_poly)
p1 = PathRegistry.coerce((p_poly_insp, emapper.attrs.machines))
- # polymorphic AliasedClass - as of #5082, for the sub entities that are
- # generated for each subclass by with_polymorphic(), use_mapper_path
- # is not True so that creating paths from the sub entities, which don't
- # by themselves encapsulate the with_polymorphic selectable, states the
- # path in terms of that plain entity. previously, this path would be
- # (e_poly, emapper.attrs.machines), but a loader strategy would never
- # match on "e_poly", it would see "emapper".
- eq_(p1.path, (emapper, emapper.attrs.machines))
+ # changes as of #5082: when a with_polymorphic is in the middle
+ # of a path, the natural path makes sure it uses the base mappers,
+ # however when it's at the root, the with_polymorphic stays in
+ # the natural path
+
+ # this behavior is the same as pre #5082, it was temporarily changed
+ # but this proved to be incorrect. The path starts on a
+ # with_polymorphic(), so a Query will "naturally" construct a path
+ # that comes from that wp.
+ eq_(p1.path, (e_poly_insp, emapper.attrs.machines))
+ eq_(p1.natural_path, (e_poly_insp, emapper.attrs.machines))
+
+ # this behavior is new as of the final version of #5082.
+ # the path starts on a normal entity and has a with_polymorphic
+ # in the middle, for this to match what Query will generate it needs
+ # to use the non aliased mappers in the natural path.
+ p2 = PathRegistry.coerce(
+ (
+ cmapper,
+ cmapper.attrs.employees,
+ p_poly_insp,
+ emapper.attrs.machines,
+ )
+ )
+ eq_(
+ p2.path,
+ (
+ cmapper,
+ cmapper.attrs.employees,
+ e_poly_insp,
+ emapper.attrs.machines,
+ ),
+ )
+ eq_(
+ p2.natural_path,
+ (
+ cmapper,
+ cmapper.attrs.employees,
+ emapper,
+ emapper.attrs.machines,
+ ),
+ )
def test_with_poly_base(self):
Person = _poly_fixtures.Person
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
