--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 4469
+
+ Improved the behavior of :func:`.orm.with_polymorphic` in conjunction with
+ loader options, in particular wildcard operations as well as
+ :func:`.orm.load_only`. The polymorphic object will be more accurately
+ targeted so that column-level options on the entity will correctly take
+ effect.The issue is a continuation of the same kinds of things fixed in
+ :ticket:`4468`.
+
def set(self, attributes, key, value):
log.debug("set '%s' on path '%s' to '%s'", key, self, value)
- attributes[(key, self.path)] = value
+ attributes[(key, self.natural_path)] = value
def setdefault(self, attributes, key, value):
log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value)
- attributes.setdefault((key, self.path), value)
+ attributes.setdefault((key, self.natural_path), value)
def get(self, attributes, key, value=None):
- key = (key, self.path)
+ key = (key, self.natural_path)
if key in attributes:
return attributes[key]
else:
"""
- path = ()
+ path = natural_path = ()
has_entity = False
is_aliased_class = False
is_root = True
self.token = token
self.parent = parent
self.path = parent.path + (token,)
+ self.natural_path = parent.natural_path + (token,)
has_entity = False
if not self.parent.is_aliased_class and not self.parent.is_root:
for ent in self.parent.mapper.iterate_to_root():
yield TokenRegistry(self.parent.parent[ent], self.token)
+ elif (
+ self.parent.is_aliased_class
+ and self.parent.entity._is_with_polymorphic
+ ):
+ yield self
+ for ent in self.parent.entity._with_polymorphic_entities:
+ yield TokenRegistry(self.parent.parent[ent], self.token)
else:
yield self
self.prop = prop
self.parent = parent
self.path = parent.path + (prop,)
+ self.natural_path = parent.natural_path + (prop,)
self._wildcard_path_loader_key = (
"loader",
self.is_aliased_class = entity.is_aliased_class
self.entity = entity
self.path = parent.path + (entity,)
+
+ # the "natural path" is the path that we get when Query is traversing
+ # from the lead entities into the various relationships; it corresponds
+ # to the structure of mappers and relationships. when we are given a
+ # path that comes from loader options, as of 1.3 it can have ac-hoc
+ # with_polymorphic() and other AliasedInsp objects inside of it, which
+ # are usually not present in mappings. So here we track both the
+ # "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:
+ if entity.mapper.isa(parent.natural_path[-1].entity):
+ self.natural_path = parent.natural_path + (entity.mapper,)
+ else:
+ self.natural_path = parent.natural_path + (
+ parent.natural_path[-1].entity,
+ )
+ else:
+ self.natural_path = self.path
self.entity_path = self
@property
path = path[attr]
elif _is_mapped_class(attr):
- if not attr.common_parent(path.mapper):
+ if not orm_util._entity_corresponds_to_use_path_impl(
+ attr.parent, path[-1]
+ ):
if raiseerr:
raise sa_exc.ArgumentError(
"Attribute '%s' does not "
else:
prop = found_property = attr.property
- if not prop.parent.common_parent(path.mapper):
+ if not orm_util._entity_corresponds_to_use_path_impl(
+ attr.parent, path[-1]
+ ):
if raiseerr:
raise sa_exc.ArgumentError(
- "Attribute '%s' does not "
- "link from element '%s'" % (attr, path.entity)
+ 'Attribute "%s" does not '
+ 'link from element "%s"' % (attr, path.entity)
)
else:
return None
_existing_alias=existing,
)
ext_info = inspect(ac)
- elif not ext_info.with_polymorphic_mappers:
- ext_info = orm_util.AliasedInsp(
- ext_info.entity,
- ext_info.mapper.base_mapper,
- ext_info.selectable,
- ext_info.name,
- ext_info.with_polymorphic_mappers or [ext_info.mapper],
- ext_info.polymorphic_on,
- ext_info._base_alias,
- ext_info._use_mapper_path,
- ext_info._adapt_on_names,
- ext_info.represents_outer_join,
- )
path.entity_path[prop].set(
self.context, "path_with_polymorphic", ext_info
)
- # the path here will go into the context dictionary and
- # needs to match up to how the class graph is traversed.
- # so we can't put an AliasedInsp in the path here, needs
- # to be the base mapper.
- path = path[prop][ext_info.mapper]
-
- # but, we need to know what the original of_type()
- # argument is for cache key purposes. so....store that too.
- # it might be better for "path" to really represent,
- # "the path", but trying to keep the impact of the cache
- # key feature localized for now
+ path = path[prop][ext_info]
+
self._of_type = of_type_info
+
else:
path = path[prop]
self.persist_selectable
) = self.local_table = selectable
self.name = name
- self.with_polymorphic_mappers = with_polymorphic_mappers
self.polymorphic_on = polymorphic_on
self._base_alias = _base_alias or self
self._use_mapper_path = _use_mapper_path
self.represents_outer_join = represents_outer_join
+ if with_polymorphic_mappers:
+ self._is_with_polymorphic = True
+ self.with_polymorphic_mappers = with_polymorphic_mappers
+ self._with_polymorphic_entities = []
+ for poly in self.with_polymorphic_mappers:
+ if poly is not mapper:
+ ent = AliasedClass(
+ poly.class_,
+ selectable,
+ base_alias=self,
+ adapt_on_names=adapt_on_names,
+ use_mapper_path=_use_mapper_path,
+ )
+
+ setattr(self.entity, poly.class_.__name__, ent)
+ self._with_polymorphic_entities.append(ent._aliased_insp)
+
+ else:
+ self._is_with_polymorphic = False
+ self.with_polymorphic_mappers = [mapper]
+
self._adapter = sql_util.ColumnAdapter(
selectable,
equivalents=mapper._equivalent_columns,
self._adapt_on_names = adapt_on_names
self._target = mapper.class_
- for poly in self.with_polymorphic_mappers:
- if poly is not mapper:
- setattr(
- self.entity,
- poly.class_.__name__,
- AliasedClass(
- poly.class_,
- selectable,
- base_alias=self,
- adapt_on_names=adapt_on_names,
- use_mapper_path=_use_mapper_path,
- ),
- )
-
is_aliased_class = True
"always returns True"
)
def __str__(self):
- return "aliased(%s)" % (self._target.__name__,)
+ if self._is_with_polymorphic:
+ return "with_polymorphic(%s, [%s])" % (
+ self._target.__name__,
+ ", ".join(
+ mp.class_.__name__
+ for mp in self.with_polymorphic_mappers
+ if mp is not self.mapper
+ ),
+ )
+ else:
+ return "aliased(%s)" % (self._target.__name__,)
inspection._inspects(AliasedClass)(lambda target: target._aliased_insp)
return entity.common_parent(given)
+def _entity_corresponds_to_use_path_impl(given, entity):
+ """determine if 'given' corresponds to 'entity', in terms
+ of a path of loader options where a mapped attribute is taken to
+ be a member of a parent entity.
+
+ e.g.::
+
+ someoption(A).someoption(A.b) # -> fn(A, A) -> True
+ someoption(A).someoption(C.d) # -> fn(A, C) -> False
+
+ a1 = aliased(A)
+ someoption(a1).someoption(A.b) # -> fn(a1, A) -> False
+ someoption(a1).someoption(a1.b) # -> fn(a1, a1) -> True
+
+ wp = with_polymorphic(A, [A1, A2])
+ someoption(wp).someoption(A1.foo) # -> fn(wp, A1) -> False
+ someoption(wp).someoption(wp.A1.foo) # -> fn(wp, wp.A1) -> True
+
+
+ """
+ if given.is_aliased_class:
+ return (
+ entity.is_aliased_class
+ and not entity._use_mapper_path
+ and given is entity
+ or given in entity._with_polymorphic_entities
+ )
+ elif not entity.is_aliased_class:
+ return given.common_parent(entity.mapper)
+ else:
+ return (
+ entity._use_mapper_path
+ and given in entity.with_polymorphic_mappers
+ )
+
+
def _entity_isa(given, mapper):
"""determine if 'given' "is a" mapper, in terms of the given
would load rows of type 'mapper'.
Column("foo", Integer),
Column("bar", Integer),
)
- m1 = mapper(A, a)
+ b = Table(
+ "b",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("a_id", ForeignKey("a.id")),
+ )
+ m1 = mapper(A, a, properties={"bs": relationship(B)})
+ m2 = mapper(B, b)
@profile_memory()
def go():
ma = sa.inspect(aliased(A))
- m1._path_registry[m1.attrs.foo][ma][m1.attrs.bar]
+ m1._path_registry[m1.attrs.bs][ma][m1.attrs.bar]
go()
clear_mappers()
q = session.query(poly).options(
joinedload(poly.Sub1.links)
.joinedload(Link.child.of_type(Sub1))
- .joinedload(poly.Sub1.links)
+ .joinedload(Sub1.links)
)
self.assert_compile(
q,
q = session.query(poly).options(
joinedload(poly.Sub1.links, innerjoin=True)
.joinedload(Link.child.of_type(Sub1), innerjoin=True)
- .joinedload(poly.Sub1.links, innerjoin=True)
+ .joinedload(Sub1.links, innerjoin=True)
)
self.assert_compile(
q,
"ORDER BY people.person_id",
)
+ def test_load_only_from_with_polymorphic(self):
+ s = Session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ 'Mapped attribute "Manager.status" does not apply to any of the '
+ "root entities in this query, e.g. "
+ r"with_polymorphic\(Person, \[Manager\]\).",
+ s.query(wp).options,
+ load_only(Manager.status),
+ )
+
+ q = s.query(wp).options(load_only(wp.Manager.status))
+ self.assert_compile(
+ q,
+ "SELECT people_1.person_id AS people_1_person_id, "
+ "people_1.type AS people_1_type, "
+ "managers_1.person_id AS managers_1_person_id, "
+ "managers_1.status AS managers_1_status "
+ "FROM people AS people_1 "
+ "LEFT OUTER JOIN managers AS managers_1 "
+ "ON people_1.person_id = managers_1.person_id",
+ )
+
+ def test_load_only_of_type_with_polymorphic(self):
+ s = Session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ # needs to be explicit, we don't currently dig onto all the
+ # sub-entities in the wp
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Can\'t find property named "status" on '
+ r"with_polymorphic\(Person, \[Manager\]\) in this Query.",
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(wp)).load_only("status"),
+ )
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ 'Attribute "Manager.status" does not link from element '
+ r'"with_polymorphic\(Person, \[Manager\]\)"',
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(wp)).load_only(
+ Manager.status
+ ),
+ )
+
+ self.assert_compile(
+ s.query(Company).options(
+ joinedload(Company.employees.of_type(wp)).load_only(
+ wp.Manager.status
+ )
+ ),
+ # should at least not have manager_name in it
+ "SELECT companies.company_id AS companies_company_id, "
+ "companies.name AS companies_name, "
+ "people_1.person_id AS people_1_person_id, "
+ "people_1.type AS people_1_type, "
+ "managers_1.person_id AS managers_1_person_id, "
+ "managers_1.status AS managers_1_status "
+ "FROM companies LEFT OUTER JOIN "
+ "(people AS people_1 LEFT OUTER JOIN managers AS managers_1 "
+ "ON people_1.person_id = managers_1.person_id) "
+ "ON companies.company_id = people_1.company_id "
+ "ORDER BY people_1.person_id",
+ )
+
class WithExpressionTest(fixtures.DeclarativeMappedTest):
@classmethod
from sqlalchemy.orm import strategy_options
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import util as orm_util
+from sqlalchemy.orm import with_polymorphic
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.assertions import assert_raises_message
from sqlalchemy.testing.assertions import eq_
from test.orm import _fixtures
+from .inheritance._poly_fixtures import _Polymorphic
+from .inheritance._poly_fixtures import Company
+from .inheritance._poly_fixtures import Engineer
+from .inheritance._poly_fixtures import Manager
+from .inheritance._poly_fixtures import Person
class QueryTest(_fixtures.FixtureTest):
)
+class OptionsNoPropTestInh(_Polymorphic):
+ def test_missing_attr_wpoly_subclasss(self):
+ s = Session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Mapped attribute "Manager.status" does not apply to any of '
+ r"the root entities in this query, e.g. "
+ r"with_polymorphic\(Person, \[Manager\]\).",
+ s.query(wp).options,
+ load_only(Manager.status),
+ )
+
+ def test_missing_attr_of_type_subclass(self):
+ s = Session()
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Attribute "Manager.manager_name" does not link from element '
+ r'"with_polymorphic\(Person, \[Engineer\]\)"',
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(Engineer)).load_only(
+ Manager.manager_name
+ ),
+ )
+
+ def test_missing_attr_of_type_subclass_name_matches(self):
+ s = Session()
+
+ # the name "status" is present on Engineer also, make sure
+ # that doesn't get mixed up here
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Attribute "Manager.status" does not link from element '
+ r'"with_polymorphic\(Person, \[Engineer\]\)"',
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(Engineer)).load_only(
+ Manager.status
+ ),
+ )
+
+ def test_missing_str_attr_of_type_subclass(self):
+ s = Session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Can\'t find property named "manager_name" on '
+ "mapped class Engineer->engineers in this Query.",
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(Engineer)).load_only(
+ "manager_name"
+ ),
+ )
+
+ def test_missing_attr_of_type_wpoly_subclass(self):
+ s = Session()
+
+ wp = with_polymorphic(Person, [Manager], flat=True)
+
+ assert_raises_message(
+ sa.exc.ArgumentError,
+ r'Attribute "Manager.manager_name" does not link from '
+ r'element "with_polymorphic\(Person, \[Manager\]\)"',
+ s.query(Company).options,
+ joinedload(Company.employees.of_type(wp)).load_only(
+ Manager.manager_name
+ ),
+ )
+
+
class PickleTest(PathTest, QueryTest):
def _option_fixture(self, *arg):
return strategy_options._UnboundLoad._from_keys(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
