Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/path_registry.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7"""Path tracking utilities, representing mapper graph traversals."""
9from __future__ import annotations
11from functools import reduce
12from itertools import chain
13import logging
14import operator
15from typing import Any
16from typing import cast
17from typing import Dict
18from typing import Final
19from typing import Iterator
20from typing import List
21from typing import Literal
22from typing import Mapping
23from typing import Optional
24from typing import overload
25from typing import Sequence
26from typing import Tuple
27from typing import TYPE_CHECKING
28from typing import Union
30from . import base as orm_base
31from ._typing import insp_is_mapper_property
32from .. import exc
33from .. import inspection
34from .. import util
35from ..sql import visitors
36from ..sql.cache_key import HasCacheKey
38if TYPE_CHECKING:
39 from typing import TypeGuard
41 from ._typing import _InternalEntityType
42 from .interfaces import StrategizedProperty
43 from .mapper import Mapper
44 from .relationships import RelationshipProperty
45 from .util import AliasedInsp
46 from ..sql.cache_key import _CacheKeyTraversalType
47 from ..sql.elements import BindParameter
48 from ..sql.visitors import anon_map
49 from ..util.typing import _LiteralStar
51 def is_root(path: PathRegistry) -> TypeGuard[RootRegistry]: ...
53 def is_entity(
54 path: PathRegistry,
55 ) -> TypeGuard[_AbstractEntityRegistry]: ...
57else:
58 is_root = operator.attrgetter("is_root")
59 is_entity = operator.attrgetter("is_entity")
62_SerializedPath = List[Any]
63_StrPathToken = str
64_PathElementType = Union[
65 _StrPathToken, "_InternalEntityType[Any]", "StrategizedProperty[Any]"
66]
68# the representation is in fact
69# a tuple with alternating:
70# [_InternalEntityType[Any], Union[str, StrategizedProperty[Any]],
71# _InternalEntityType[Any], Union[str, StrategizedProperty[Any]], ...]
72# this might someday be a tuple of 2-tuples instead, but paths can be
73# chopped at odd intervals as well so this is less flexible
74_PathRepresentation = Tuple[_PathElementType, ...]
76# NOTE: these names are weird since the array is 0-indexed,
77# the "_Odd" entries are at 0, 2, 4, etc
78_OddPathRepresentation = Sequence["_InternalEntityType[Any]"]
79_EvenPathRepresentation = Sequence[Union["StrategizedProperty[Any]", str]]
82log = logging.getLogger(__name__)
85def _unreduce_path(path: _SerializedPath) -> PathRegistry:
86 return PathRegistry.deserialize(path)
89_WILDCARD_TOKEN: _LiteralStar = "*"
90_DEFAULT_TOKEN = "_sa_default"
92_RELATIONSHIP_TOKEN: Final[Literal["relationship"]] = "relationship"
93_COLUMN_TOKEN: Final[Literal["column"]] = "column"
95_UNPREFIXED_TOKENS = frozenset([_WILDCARD_TOKEN, _DEFAULT_TOKEN])
96"""the wildcard strings accepted from the user in a loader option.
98these are prefixed with the target property's ``strategy_wildcard_key`` to
99form the tokens in :data:`._PATH_TOKENS`, and are not themselves valid as an
100element of a path.
102"""
104_PATH_TOKENS = frozenset(
105 f"{wildcard_key}:{suffix}"
106 for wildcard_key in (_RELATIONSHIP_TOKEN, _COLUMN_TOKEN)
107 for suffix in (_WILDCARD_TOKEN, _DEFAULT_TOKEN)
108)
109"""the complete set of tokens which may appear within a path.
111:attr:`.PathToken._intern` is populated from this collection at module import
112time, so that a token is present in every process, including one which has
113not yet run any query.
115"""
117_ACCEPTED_TOKENS = _UNPREFIXED_TOKENS | _PATH_TOKENS
118"""every string a loader option may accept in place of an attribute name.
120this is the union of the bare wildcards the user writes and the prefixed
121tokens which the loader option internals hand back to themselves; only the
122latter may appear in a path.
124"""
127@inspection._self_inspects
128class PathRegistry(HasCacheKey):
129 """Represent query load paths and registry functions.
131 Basically represents structures like:
133 (<User mapper>, "orders", <Order mapper>, "items", <Item mapper>)
135 These structures are generated by things like
136 query options (joinedload(), subqueryload(), etc.) and are
137 used to compose keys stored in the query._attributes dictionary
138 for various options.
140 They are then re-composed at query compile/result row time as
141 the query is formed and as rows are fetched, where they again
142 serve to compose keys to look up options in the context.attributes
143 dictionary, which is copied from query._attributes.
145 The path structure has a limited amount of caching, where each
146 "root" ultimately pulls from a fixed registry associated with
147 the first mapper, that also contains elements for each of its
148 property keys. However paths longer than two elements, which
149 are the exception rather than the rule, are generated on an
150 as-needed basis.
152 """
154 __slots__ = ()
156 is_token = False
157 is_root = False
158 has_entity = False
159 is_property = False
160 is_entity = False
162 is_unnatural: bool
164 path: _PathRepresentation
165 natural_path: _PathRepresentation
166 parent: Optional[PathRegistry]
167 root: RootRegistry
169 _cache_key_traversal: _CacheKeyTraversalType = [
170 ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key_list)
171 ]
173 def __eq__(self, other: Any) -> bool:
174 try:
175 return other is not None and self.path == other._path_for_compare
176 except AttributeError:
177 util.warn(
178 "Comparison of PathRegistry to %r is not supported"
179 % (type(other))
180 )
181 return False
183 def __ne__(self, other: Any) -> bool:
184 try:
185 return other is None or self.path != other._path_for_compare
186 except AttributeError:
187 util.warn(
188 "Comparison of PathRegistry to %r is not supported"
189 % (type(other))
190 )
191 return True
193 @property
194 def _path_for_compare(self) -> Optional[_PathRepresentation]:
195 return self.path
197 def odd_element(self, index: int) -> _InternalEntityType[Any]:
198 return self.path[index] # type: ignore[return-value]
200 def set(self, attributes: Dict[Any, Any], key: Any, value: Any) -> None:
201 log.debug("set '%s' on path '%s' to '%s'", key, self, value)
202 attributes[(key, self.natural_path)] = value
204 def setdefault(
205 self, attributes: Dict[Any, Any], key: Any, value: Any
206 ) -> None:
207 log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value)
208 attributes.setdefault((key, self.natural_path), value)
210 def get(
211 self, attributes: Dict[Any, Any], key: Any, value: Optional[Any] = None
212 ) -> Any:
213 key = (key, self.natural_path)
214 if key in attributes:
215 return attributes[key]
216 else:
217 return value
219 def __len__(self) -> int:
220 return len(self.path)
222 def __hash__(self) -> int:
223 return id(self)
225 @overload
226 def __getitem__(self, entity: _StrPathToken) -> _TokenRegistry: ...
228 @overload
229 def __getitem__(self, entity: int) -> _PathElementType: ...
231 @overload
232 def __getitem__(self, entity: slice) -> _PathRepresentation: ...
234 @overload
235 def __getitem__(
236 self, entity: _InternalEntityType[Any]
237 ) -> _AbstractEntityRegistry: ...
239 @overload
240 def __getitem__(
241 self, entity: StrategizedProperty[Any]
242 ) -> _PropRegistry: ...
244 def __getitem__(
245 self,
246 entity: Union[
247 _StrPathToken,
248 int,
249 slice,
250 _InternalEntityType[Any],
251 StrategizedProperty[Any],
252 ],
253 ) -> Union[
254 _TokenRegistry,
255 _PathElementType,
256 _PathRepresentation,
257 _PropRegistry,
258 _AbstractEntityRegistry,
259 ]:
260 raise NotImplementedError()
262 # TODO: what are we using this for?
263 @property
264 def length(self) -> int:
265 return len(self.path)
267 def pairs(
268 self,
269 ) -> Iterator[
270 Tuple[_InternalEntityType[Any], Union[str, StrategizedProperty[Any]]]
271 ]:
272 odd_path = cast(_OddPathRepresentation, self.path)
273 even_path = cast(_EvenPathRepresentation, odd_path)
274 for i in range(0, len(odd_path), 2):
275 yield odd_path[i], even_path[i + 1]
277 def contains_mapper(self, mapper: Mapper[Any]) -> bool:
278 _m_path = cast(_OddPathRepresentation, self.path)
279 for path_mapper in [_m_path[i] for i in range(0, len(_m_path), 2)]:
280 if path_mapper.mapper.isa(mapper):
281 return True
282 else:
283 return False
285 def contains(self, attributes: Dict[Any, Any], key: Any) -> bool:
286 return (key, self.path) in attributes
288 def __reduce__(self) -> Any:
289 return _unreduce_path, (self.serialize(),)
291 @classmethod
292 def _serialize_path(cls, path: _PathRepresentation) -> _SerializedPath:
293 _m_path = cast(_OddPathRepresentation, path)
294 _p_path = cast(_EvenPathRepresentation, path)
296 return list(
297 zip(
298 tuple(
299 m.class_ if (m.is_mapper or m.is_aliased_class) else str(m)
300 for m in [_m_path[i] for i in range(0, len(_m_path), 2)]
301 ),
302 tuple(
303 p.key if insp_is_mapper_property(p) else str(p)
304 for p in [_p_path[i] for i in range(1, len(_p_path), 2)]
305 )
306 + (None,),
307 )
308 )
310 @classmethod
311 def _deserialize_path(cls, path: _SerializedPath) -> _PathRepresentation:
312 def _deserialize_mapper_token(mcls: Any) -> Any:
313 return (
314 # note: we likely dont want configure=True here however
315 # this is maintained at the moment for backwards compatibility
316 orm_base._inspect_mapped_class(mcls, configure=True)
317 if mcls not in PathToken._intern
318 else PathToken._intern[mcls]
319 )
321 def _deserialize_key_token(mcls: Any, key: Any) -> Any:
322 if key is None:
323 return None
324 elif key in PathToken._intern:
325 return PathToken._intern[key]
326 else:
327 mp = orm_base._inspect_mapped_class(mcls, configure=True)
328 assert mp is not None
329 return mp.attrs[key]
331 p = tuple(
332 chain(
333 *[
334 (
335 _deserialize_mapper_token(mcls),
336 _deserialize_key_token(mcls, key),
337 )
338 for mcls, key in path
339 ]
340 )
341 )
342 if p and p[-1] is None:
343 p = p[0:-1]
344 return p
346 def serialize(self) -> _SerializedPath:
347 path = self.path
348 return self._serialize_path(path)
350 @classmethod
351 def deserialize(cls, path: _SerializedPath) -> PathRegistry:
352 assert path is not None
353 p = cls._deserialize_path(path)
354 return cls.coerce(p)
356 @overload
357 @classmethod
358 def per_mapper(cls, mapper: Mapper[Any]) -> _CachingEntityRegistry: ...
360 @overload
361 @classmethod
362 def per_mapper(cls, mapper: AliasedInsp[Any]) -> _SlotsEntityRegistry: ...
364 @classmethod
365 def per_mapper(
366 cls, mapper: _InternalEntityType[Any]
367 ) -> _AbstractEntityRegistry:
368 if mapper.is_mapper:
369 return _CachingEntityRegistry(cls.root, mapper)
370 else:
371 return _SlotsEntityRegistry(cls.root, mapper)
373 @classmethod
374 def coerce(cls, raw: _PathRepresentation) -> PathRegistry:
375 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
376 return prev[next_]
378 # can't quite get mypy to appreciate this one :)
379 return reduce(_red, raw, cls.root) # type: ignore[arg-type]
381 def __add__(self, other: PathRegistry) -> PathRegistry:
382 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
383 return prev[next_]
385 return reduce(_red, other.path, self)
387 def __str__(self) -> str:
388 return f"ORM Path[{' -> '.join(str(elem) for elem in self.path)}]"
390 def __repr__(self) -> str:
391 return f"{self.__class__.__name__}({self.path!r})"
393 def path_string(self) -> str:
394 """Return a user-facing string representation of this path,
395 e.g. ``"User.orders -> Order.items"``.
397 """
399 raw = self.path
400 parts = []
401 lraw = len(raw)
402 for i in range(0, lraw - 1, 2):
403 entity = raw[i]
404 prop = raw[i + 1]
405 prop_key = getattr(prop, "key", str(prop))
407 if (
408 i < lraw - 2
409 and cast(
410 "_InternalEntityType[Any]", raw[i + 2]
411 ).is_aliased_class
412 ):
413 parts.append(
414 f"{orm_base.entity_str(entity)}.{prop_key}."
415 f"of_type({orm_base.entity_str(raw[i + 2])})"
416 )
417 else:
418 parts.append(f"{orm_base.entity_str(entity)}.{prop_key}")
420 return (
421 " -> ".join(parts)
422 if parts
423 else orm_base.entity_str(self.path[0]) if self.path else ""
424 )
427class _CreatesToken(PathRegistry):
428 __slots__ = ()
430 is_aliased_class: bool
431 is_root: bool
433 def token(self, token: _StrPathToken) -> _TokenRegistry:
434 if token not in PathToken._intern:
435 raise exc.ArgumentError(f"invalid token: {token}")
436 elif token.endswith(f":{_WILDCARD_TOKEN}"):
437 return _TokenRegistry(self, token)
438 else:
439 return _TokenRegistry(self.root, token)
442class RootRegistry(_CreatesToken):
443 """Root registry, defers to mappers so that
444 paths are maintained per-root-mapper.
446 """
448 __slots__ = ()
450 inherit_cache = True
452 path = natural_path = ()
453 has_entity = False
454 is_aliased_class = False
455 is_root = True
456 is_unnatural = False
458 def _getitem(
459 self, entity: Any
460 ) -> Union[_TokenRegistry, _AbstractEntityRegistry]:
461 if entity in PathToken._intern:
462 if TYPE_CHECKING:
463 assert isinstance(entity, _StrPathToken)
464 return _TokenRegistry(self, PathToken._intern[entity])
465 else:
466 try:
467 return entity._path_registry # type: ignore[no-any-return]
468 except AttributeError:
469 raise IndexError(
470 f"invalid argument for RootRegistry.__getitem__: {entity}"
471 )
473 def _truncate_recursive(self) -> RootRegistry:
474 return self
476 if not TYPE_CHECKING:
477 __getitem__ = _getitem
480PathRegistry.root = RootRegistry()
483class PathToken(orm_base.InspectionAttr, HasCacheKey, str):
484 """cacheable string token"""
486 _intern: Mapping[str, PathToken]
487 """the :class:`.PathToken` for each of :data:`._PATH_TOKENS`.
489 this collection is fully populated below at module import time and is
490 never added to afterwards; it's typed as :class:`.Mapping` so that a
491 mutation is flagged by type checkers.
492 :meth:`.PathRegistry._deserialize_path` relies on it being complete,
493 distinguishing a token from a mapped attribute key by testing
494 membership here.
496 """
498 def _gen_cache_key(
499 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
500 ) -> Tuple[Any, ...]:
501 return (str(self),)
503 @property
504 def _path_for_compare(self) -> Optional[_PathRepresentation]:
505 return None
508PathToken._intern = {token: PathToken(token) for token in _PATH_TOKENS}
511class _TokenRegistry(PathRegistry):
512 __slots__ = ("token", "parent", "path", "natural_path")
514 inherit_cache = True
516 token: _StrPathToken
517 parent: _CreatesToken
519 def __init__(self, parent: _CreatesToken, token: _StrPathToken):
520 token = PathToken._intern[token]
522 self.token = token
523 self.parent = parent
524 self.path = parent.path + (token,)
525 self.natural_path = parent.natural_path + (token,)
527 has_entity = False
529 is_token = True
531 def generate_for_superclasses(self) -> Iterator[PathRegistry]:
532 # NOTE: this method is no longer used. consider removal
533 parent = self.parent
534 if is_root(parent):
535 yield self
536 return
538 if TYPE_CHECKING:
539 assert isinstance(parent, _AbstractEntityRegistry)
540 if not parent.is_aliased_class:
541 for mp_ent in parent.mapper.iterate_to_root():
542 yield _TokenRegistry(parent.parent[mp_ent], self.token)
543 elif (
544 parent.is_aliased_class
545 and cast(
546 "AliasedInsp[Any]",
547 parent.entity,
548 )._is_with_polymorphic
549 ):
550 yield self
551 for ent in cast(
552 "AliasedInsp[Any]", parent.entity
553 )._with_polymorphic_entities:
554 yield _TokenRegistry(parent.parent[ent], self.token)
555 else:
556 yield self
558 def _generate_natural_for_superclasses(
559 self,
560 ) -> Iterator[_PathRepresentation]:
561 parent = self.parent
562 if is_root(parent):
563 yield self.natural_path
564 return
566 if TYPE_CHECKING:
567 assert isinstance(parent, _AbstractEntityRegistry)
568 for mp_ent in parent.mapper.iterate_to_root():
569 yield _TokenRegistry(
570 parent.parent[mp_ent], self.token
571 ).natural_path
572 if (
573 parent.is_aliased_class
574 and cast(
575 "AliasedInsp[Any]",
576 parent.entity,
577 )._is_with_polymorphic
578 ):
579 yield self.natural_path
580 for ent in cast(
581 "AliasedInsp[Any]", parent.entity
582 )._with_polymorphic_entities:
583 yield (
584 _TokenRegistry(parent.parent[ent], self.token).natural_path
585 )
586 else:
587 yield self.natural_path
589 def _getitem(self, entity: Any) -> Any:
590 try:
591 return self.path[entity]
592 except TypeError as err:
593 raise IndexError(f"{entity}") from err
595 if not TYPE_CHECKING:
596 __getitem__ = _getitem
599class _PropRegistry(PathRegistry):
600 __slots__ = (
601 "prop",
602 "parent",
603 "path",
604 "natural_path",
605 "has_entity",
606 "entity",
607 "mapper",
608 "_wildcard_path_loader_key",
609 "_default_path_loader_key",
610 "_loader_key",
611 "is_unnatural",
612 )
613 inherit_cache = True
614 is_property = True
616 prop: StrategizedProperty[Any]
617 mapper: Optional[Mapper[Any]]
618 entity: Optional[_InternalEntityType[Any]]
619 parent: _AbstractEntityRegistry
621 def __init__(
622 self, parent: _AbstractEntityRegistry, prop: StrategizedProperty[Any]
623 ):
625 # restate this path in terms of the
626 # given StrategizedProperty's parent.
627 insp = cast("_InternalEntityType[Any]", parent[-1])
628 natural_parent: _AbstractEntityRegistry = parent
630 # inherit "is_unnatural" from the parent
631 self.is_unnatural = parent.parent.is_unnatural or bool(
632 parent.mapper.inherits
633 )
635 if not insp.is_aliased_class or insp._use_mapper_path: # type: ignore[union-attr] # noqa: E501
636 parent = natural_parent = parent.parent[prop.parent]
637 elif (
638 insp.is_aliased_class
639 and insp.with_polymorphic_mappers
640 and prop.parent in insp.with_polymorphic_mappers
641 ):
642 subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent) # type: ignore[union-attr] # noqa: E501
643 parent = parent.parent[subclass_entity]
645 # when building a path where with_polymorphic() is in use,
646 # special logic to determine the "natural path" when subclass
647 # entities are used.
648 #
649 # here we are trying to distinguish between a path that starts
650 # on a with_polymorphic entity vs. one that starts on a
651 # normal entity that introduces a with_polymorphic() in the
652 # middle using of_type():
653 #
654 # # as in test_polymorphic_rel->
655 # # test_subqueryload_on_subclass_uses_path_correctly
656 # wp = with_polymorphic(RegularEntity, "*")
657 # sess.query(wp).options(someload(wp.SomeSubEntity.foos))
658 #
659 # vs
660 #
661 # # as in test_relationship->JoinedloadWPolyOfTypeContinued
662 # wp = with_polymorphic(SomeFoo, "*")
663 # sess.query(RegularEntity).options(
664 # someload(RegularEntity.foos.of_type(wp))
665 # .someload(wp.SubFoo.bar)
666 # )
667 #
668 # in the former case, the Query as it generates a path that we
669 # want to match will be in terms of the with_polymorphic at the
670 # beginning. in the latter case, Query will generate simple
671 # paths that don't know about this with_polymorphic, so we must
672 # use a separate natural path.
673 #
674 #
675 if parent.parent:
676 natural_parent = parent.parent[subclass_entity.mapper]
677 self.is_unnatural = True
678 else:
679 natural_parent = parent
680 elif (
681 natural_parent.parent
682 and insp.is_aliased_class
683 and prop.parent # this should always be the case here
684 is not insp.mapper
685 and insp.mapper.isa(prop.parent)
686 ):
687 natural_parent = parent.parent[prop.parent]
689 self.prop = prop
690 self.parent = parent
691 self.path = parent.path + (prop,)
692 self.natural_path = natural_parent.natural_path + (prop,)
694 self.has_entity = prop._links_to_entity
695 if prop._is_relationship:
696 if TYPE_CHECKING:
697 assert isinstance(prop, RelationshipProperty)
698 self.entity = prop.entity
699 self.mapper = prop.mapper
700 else:
701 self.entity = None
702 self.mapper = None
704 self._wildcard_path_loader_key = (
705 "loader",
706 parent.natural_path + self.prop._wildcard_token,
707 )
708 self._default_path_loader_key = self.prop._default_path_loader_key
709 self._loader_key = ("loader", self.natural_path)
711 def _truncate_recursive(self) -> _PropRegistry:
712 earliest = None
713 for i, token in enumerate(reversed(self.path[:-1])):
714 if token is self.prop:
715 earliest = i
717 if earliest is None:
718 return self
719 else:
720 return self.coerce(self.path[0 : -(earliest + 1)]) # type: ignore[return-value] # noqa: E501
722 @property
723 def entity_path(self) -> _AbstractEntityRegistry:
724 assert self.entity is not None
725 return self[self.entity]
727 def _getitem(
728 self, entity: Union[int, slice, _InternalEntityType[Any]]
729 ) -> Union[_AbstractEntityRegistry, _PathElementType, _PathRepresentation]:
730 if isinstance(entity, (int, slice)):
731 return self.path[entity]
732 else:
733 return _SlotsEntityRegistry(self, entity)
735 if not TYPE_CHECKING:
736 __getitem__ = _getitem
739class _AbstractEntityRegistry(_CreatesToken):
740 __slots__ = (
741 "key",
742 "parent",
743 "is_aliased_class",
744 "path",
745 "entity",
746 "natural_path",
747 )
749 has_entity = True
750 is_entity = True
752 parent: Union[RootRegistry, _PropRegistry]
753 key: _InternalEntityType[Any]
754 entity: _InternalEntityType[Any]
755 is_aliased_class: bool
757 def __init__(
758 self,
759 parent: Union[RootRegistry, _PropRegistry],
760 entity: _InternalEntityType[Any],
761 ):
762 self.key = entity
763 self.parent = parent
764 self.is_aliased_class = entity.is_aliased_class
765 self.entity = entity
766 self.path = parent.path + (entity,)
768 # the "natural path" is the path that we get when Query is traversing
769 # from the lead entities into the various relationships; it corresponds
770 # to the structure of mappers and relationships. when we are given a
771 # path that comes from loader options, as of 1.3 it can have ac-hoc
772 # with_polymorphic() and other AliasedInsp objects inside of it, which
773 # are usually not present in mappings. So here we track both the
774 # "enhanced" path in self.path and the "natural" path that doesn't
775 # include those objects so these two traversals can be matched up.
777 # the test here for "(self.is_aliased_class or parent.is_unnatural)"
778 # are to avoid the more expensive conditional logic that follows if we
779 # know we don't have to do it. This conditional can just as well be
780 # "if parent.path:", it just is more function calls.
781 #
782 # This is basically the only place that the "is_unnatural" flag
783 # actually changes behavior.
784 if parent.path and (self.is_aliased_class or parent.is_unnatural):
785 # this is an infrequent code path used for loader strategies that
786 # also make use of of_type() or other intricate polymorphic
787 # base/subclass combinations
788 parent_natural_entity = parent.natural_path[-1]
790 if entity.mapper.isa(
791 parent_natural_entity.mapper # type: ignore[union-attr]
792 ) or parent_natural_entity.mapper.isa( # type: ignore[union-attr]
793 entity.mapper
794 ):
795 # when the entity mapper and parent mapper are in an
796 # inheritance relationship, use entity.mapper in natural_path.
797 # First case: entity.mapper inherits from parent mapper (e.g.,
798 # accessing a subclass mapper through parent path). Second case
799 # (issue #13193): parent mapper inherits from entity.mapper
800 # (e.g., parent path has Sub(Base) but we're accessing with
801 # Base where Base.related is declared, so use Base in
802 # natural_path).
803 self.natural_path = parent.natural_path + (entity.mapper,)
804 else:
805 self.natural_path = parent.natural_path + (
806 parent_natural_entity.entity, # type: ignore[operator, union-attr] # noqa: E501
807 )
808 # it seems to make sense that since these paths get mixed up
809 # with statements that are cached or not, we should make
810 # sure the natural path is cacheable across different occurrences
811 # of equivalent AliasedClass objects. however, so far this
812 # does not seem to be needed for whatever reason.
813 # elif not parent.path and self.is_aliased_class:
814 # self.natural_path = (self.entity._generate_cache_key()[0], )
815 else:
816 self.natural_path = self.path
818 def _truncate_recursive(self) -> _AbstractEntityRegistry:
819 return self.parent._truncate_recursive()[self.entity]
821 @property
822 def root_entity(self) -> _InternalEntityType[Any]:
823 return self.odd_element(0)
825 @property
826 def entity_path(self) -> PathRegistry:
827 return self
829 @property
830 def mapper(self) -> Mapper[Any]:
831 return self.entity.mapper
833 def __bool__(self) -> bool:
834 return True
836 def _getitem(
837 self, entity: Any
838 ) -> Union[_PathElementType, _PathRepresentation, PathRegistry]:
839 if isinstance(entity, (int, slice)):
840 return self.path[entity]
841 elif entity in PathToken._intern:
842 return _TokenRegistry(self, PathToken._intern[entity])
843 else:
844 return _PropRegistry(self, entity)
846 if not TYPE_CHECKING:
847 __getitem__ = _getitem
850class _SlotsEntityRegistry(_AbstractEntityRegistry):
851 # for aliased class, return lightweight, no-cycles created
852 # version
853 inherit_cache = True
856class _ERDict(Dict[Any, Any]):
857 def __init__(self, registry: _CachingEntityRegistry):
858 self.registry = registry
860 def __missing__(self, key: Any) -> _PropRegistry:
861 self[key] = item = _PropRegistry(self.registry, key)
863 return item
866class _CachingEntityRegistry(_AbstractEntityRegistry):
867 # for long lived mapper, return dict based caching
868 # version that creates reference cycles
870 __slots__ = ("_cache",)
872 inherit_cache = True
874 def __init__(
875 self,
876 parent: Union[RootRegistry, _PropRegistry],
877 entity: _InternalEntityType[Any],
878 ):
879 super().__init__(parent, entity)
880 self._cache = _ERDict(self)
882 def pop(self, key: Any, default: Any) -> Any:
883 return self._cache.pop(key, default)
885 def _getitem(self, entity: Any) -> Any:
886 if isinstance(entity, (int, slice)):
887 return self.path[entity]
888 elif isinstance(entity, PathToken):
889 return _TokenRegistry(self, entity)
890 else:
891 return self._cache[entity]
893 if not TYPE_CHECKING:
894 __getitem__ = _getitem
897if TYPE_CHECKING:
899 def path_is_entity(
900 path: PathRegistry,
901 ) -> TypeGuard[_AbstractEntityRegistry]: ...
903 def path_is_property(path: PathRegistry) -> TypeGuard[_PropRegistry]: ...
905else:
906 path_is_entity = operator.attrgetter("is_entity")
907 path_is_property = operator.attrgetter("is_property")