codekingpro/portable-devtools
115k
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."""
8
9from __future__ import annotations
10
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 Iterator
19from typing import List
20from typing import Optional
21from typing import overload
22from typing import Sequence
23from typing import Tuple
24from typing import TYPE_CHECKING
25from typing import Union
26
27from . import base as orm_base
28from ._typing import insp_is_mapper_property
29from .. import exc
30from .. import util
31from ..sql import visitors
32from ..sql.cache_key import HasCacheKey
33
34if TYPE_CHECKING:
35 from ._typing import _InternalEntityType
36 from .interfaces import StrategizedProperty
37 from .mapper import Mapper
38 from .relationships import RelationshipProperty
39 from .util import AliasedInsp
40 from ..sql.cache_key import _CacheKeyTraversalType
41 from ..sql.elements import BindParameter
42 from ..sql.visitors import anon_map
43 from ..util.typing import _LiteralStar
44 from ..util.typing import TypeGuard
45
46 def is_root(path: PathRegistry) -> TypeGuard[RootRegistry]: ...
47
48 def is_entity(path: PathRegistry) -> TypeGuard[AbstractEntityRegistry]: ...
49
50else:
51 is_root = operator.attrgetter("is_root")
52 is_entity = operator.attrgetter("is_entity")
53
54
55_SerializedPath = List[Any]
56_StrPathToken = str
57_PathElementType = Union[
58 _StrPathToken, "_InternalEntityType[Any]", "StrategizedProperty[Any]"
59]
60
61# the representation is in fact
62# a tuple with alternating:
63# [_InternalEntityType[Any], Union[str, StrategizedProperty[Any]],
64# _InternalEntityType[Any], Union[str, StrategizedProperty[Any]], ...]
65# this might someday be a tuple of 2-tuples instead, but paths can be
66# chopped at odd intervals as well so this is less flexible
67_PathRepresentation = Tuple[_PathElementType, ...]
68
69# NOTE: these names are weird since the array is 0-indexed,
70# the "_Odd" entries are at 0, 2, 4, etc
71_OddPathRepresentation = Sequence["_InternalEntityType[Any]"]
72_EvenPathRepresentation = Sequence[Union["StrategizedProperty[Any]", str]]
73
74
75log = logging.getLogger(__name__)
76
77
78def _unreduce_path(path: _SerializedPath) -> PathRegistry:
79 return PathRegistry.deserialize(path)
80
81
82_WILDCARD_TOKEN: _LiteralStar = "*"
83_DEFAULT_TOKEN = "_sa_default"
84
85
86class PathRegistry(HasCacheKey):
87 """Represent query load paths and registry functions.
88
89 Basically represents structures like:
90
91 (<User mapper>, "orders", <Order mapper>, "items", <Item mapper>)
92
93 These structures are generated by things like
94 query options (joinedload(), subqueryload(), etc.) and are
95 used to compose keys stored in the query._attributes dictionary
96 for various options.
97
98 They are then re-composed at query compile/result row time as
99 the query is formed and as rows are fetched, where they again
100 serve to compose keys to look up options in the context.attributes
101 dictionary, which is copied from query._attributes.
102
103 The path structure has a limited amount of caching, where each
104 "root" ultimately pulls from a fixed registry associated with
105 the first mapper, that also contains elements for each of its
106 property keys. However paths longer than two elements, which
107 are the exception rather than the rule, are generated on an
108 as-needed basis.
109
110 """
111
112 __slots__ = ()
113
114 is_token = False
115 is_root = False
116 has_entity = False
117 is_property = False
118 is_entity = False
119
120 is_unnatural: bool
121
122 path: _PathRepresentation
123 natural_path: _PathRepresentation
124 parent: Optional[PathRegistry]
125 root: RootRegistry
126
127 _cache_key_traversal: _CacheKeyTraversalType = [
128 ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key_list)
129 ]
130
131 def __eq__(self, other: Any) -> bool:
132 try:
133 return other is not None and self.path == other._path_for_compare
134 except AttributeError:
135 util.warn(
136 "Comparison of PathRegistry to %r is not supported"
137 % (type(other))
138 )
139 return False
140
141 def __ne__(self, other: Any) -> bool:
142 try:
143 return other is None or self.path != other._path_for_compare
144 except AttributeError:
145 util.warn(
146 "Comparison of PathRegistry to %r is not supported"
147 % (type(other))
148 )
149 return True
150
151 @property
152 def _path_for_compare(self) -> Optional[_PathRepresentation]:
153 return self.path
154
155 def odd_element(self, index: int) -> _InternalEntityType[Any]:
156 return self.path[index] # type: ignore
157
158 def set(self, attributes: Dict[Any, Any], key: Any, value: Any) -> None:
159 log.debug("set '%s' on path '%s' to '%s'", key, self, value)
160 attributes[(key, self.natural_path)] = value
161
162 def setdefault(
163 self, attributes: Dict[Any, Any], key: Any, value: Any
164 ) -> None:
165 log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value)
166 attributes.setdefault((key, self.natural_path), value)
167
168 def get(
169 self, attributes: Dict[Any, Any], key: Any, value: Optional[Any] = None
170 ) -> Any:
171 key = (key, self.natural_path)
172 if key in attributes:
173 return attributes[key]
174 else:
175 return value
176
177 def __len__(self) -> int:
178 return len(self.path)
179
180 def __hash__(self) -> int:
181 return id(self)
182
183 @overload
184 def __getitem__(self, entity: _StrPathToken) -> TokenRegistry: ...
185
186 @overload
187 def __getitem__(self, entity: int) -> _PathElementType: ...
188
189 @overload
190 def __getitem__(self, entity: slice) -> _PathRepresentation: ...
191
192 @overload
193 def __getitem__(
194 self, entity: _InternalEntityType[Any]
195 ) -> AbstractEntityRegistry: ...
196
197 @overload
198 def __getitem__(
199 self, entity: StrategizedProperty[Any]
200 ) -> PropRegistry: ...
201
202 def __getitem__(
203 self,
204 entity: Union[
205 _StrPathToken,
206 int,
207 slice,
208 _InternalEntityType[Any],
209 StrategizedProperty[Any],
210 ],
211 ) -> Union[
212 TokenRegistry,
213 _PathElementType,
214 _PathRepresentation,
215 PropRegistry,
216 AbstractEntityRegistry,
217 ]:
218 raise NotImplementedError()
219
220 # TODO: what are we using this for?
221 @property
222 def length(self) -> int:
223 return len(self.path)
224
225 def pairs(
226 self,
227 ) -> Iterator[
228 Tuple[_InternalEntityType[Any], Union[str, StrategizedProperty[Any]]]
229 ]:
230 odd_path = cast(_OddPathRepresentation, self.path)
231 even_path = cast(_EvenPathRepresentation, odd_path)
232 for i in range(0, len(odd_path), 2):
233 yield odd_path[i], even_path[i + 1]
234
235 def contains_mapper(self, mapper: Mapper[Any]) -> bool:
236 _m_path = cast(_OddPathRepresentation, self.path)
237 for path_mapper in [_m_path[i] for i in range(0, len(_m_path), 2)]:
238 if path_mapper.mapper.isa(mapper):
239 return True
240 else:
241 return False
242
243 def contains(self, attributes: Dict[Any, Any], key: Any) -> bool:
244 return (key, self.path) in attributes
245
246 def __reduce__(self) -> Any:
247 return _unreduce_path, (self.serialize(),)
248
249 @classmethod
250 def _serialize_path(cls, path: _PathRepresentation) -> _SerializedPath:
251 _m_path = cast(_OddPathRepresentation, path)
252 _p_path = cast(_EvenPathRepresentation, path)
253
254 return list(
255 zip(
256 tuple(
257 m.class_ if (m.is_mapper or m.is_aliased_class) else str(m)
258 for m in [_m_path[i] for i in range(0, len(_m_path), 2)]
259 ),
260 tuple(
261 p.key if insp_is_mapper_property(p) else str(p)
262 for p in [_p_path[i] for i in range(1, len(_p_path), 2)]
263 )
264 + (None,),
265 )
266 )
267
268 @classmethod
269 def _deserialize_path(cls, path: _SerializedPath) -> _PathRepresentation:
270 def _deserialize_mapper_token(mcls: Any) -> Any:
271 return (
272 # note: we likely dont want configure=True here however
273 # this is maintained at the moment for backwards compatibility
274 orm_base._inspect_mapped_class(mcls, configure=True)
275 if mcls not in PathToken._intern
276 else PathToken._intern[mcls]
277 )
278
279 def _deserialize_key_token(mcls: Any, key: Any) -> Any:
280 if key is None:
281 return None
282 elif key in PathToken._intern:
283 return PathToken._intern[key]
284 else:
285 mp = orm_base._inspect_mapped_class(mcls, configure=True)
286 assert mp is not None
287 return mp.attrs[key]
288
289 p = tuple(
290 chain(
291 *[
292 (
293 _deserialize_mapper_token(mcls),
294 _deserialize_key_token(mcls, key),
295 )
296 for mcls, key in path
297 ]
298 )
299 )
300 if p and p[-1] is None:
301 p = p[0:-1]
302 return p
303
304 def serialize(self) -> _SerializedPath:
305 path = self.path
306 return self._serialize_path(path)
307
308 @classmethod
309 def deserialize(cls, path: _SerializedPath) -> PathRegistry:
310 assert path is not None
311 p = cls._deserialize_path(path)
312 return cls.coerce(p)
313
314 @overload
315 @classmethod
316 def per_mapper(cls, mapper: Mapper[Any]) -> CachingEntityRegistry: ...
317
318 @overload
319 @classmethod
320 def per_mapper(cls, mapper: AliasedInsp[Any]) -> SlotsEntityRegistry: ...
321
322 @classmethod
323 def per_mapper(
324 cls, mapper: _InternalEntityType[Any]
325 ) -> AbstractEntityRegistry:
326 if mapper.is_mapper:
327 return CachingEntityRegistry(cls.root, mapper)
328 else:
329 return SlotsEntityRegistry(cls.root, mapper)
330
331 @classmethod
332 def coerce(cls, raw: _PathRepresentation) -> PathRegistry:
333 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
334 return prev[next_]
335
336 # can't quite get mypy to appreciate this one :)
337 return reduce(_red, raw, cls.root) # type: ignore
338
339 def __add__(self, other: PathRegistry) -> PathRegistry:
340 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
341 return prev[next_]
342
343 return reduce(_red, other.path, self)
344
345 def __str__(self) -> str:
346 return f"ORM Path[{' -> '.join(str(elem) for elem in self.path)}]"
347
348 def __repr__(self) -> str:
349 return f"{self.__class__.__name__}({self.path!r})"
350
351
352class CreatesToken(PathRegistry):
353 __slots__ = ()
354
355 is_aliased_class: bool
356 is_root: bool
357
358 def token(self, token: _StrPathToken) -> TokenRegistry:
359 if token.endswith(f":{_WILDCARD_TOKEN}"):
360 return TokenRegistry(self, token)
361 elif token.endswith(f":{_DEFAULT_TOKEN}"):
362 return TokenRegistry(self.root, token)
363 else:
364 raise exc.ArgumentError(f"invalid token: {token}")
365
366
367class RootRegistry(CreatesToken):
368 """Root registry, defers to mappers so that
369 paths are maintained per-root-mapper.
370
371 """
372
373 __slots__ = ()
374
375 inherit_cache = True
376
377 path = natural_path = ()
378 has_entity = False
379 is_aliased_class = False
380 is_root = True
381 is_unnatural = False
382
383 def _getitem(
384 self, entity: Any
385 ) -> Union[TokenRegistry, AbstractEntityRegistry]:
386 if entity in PathToken._intern:
387 if TYPE_CHECKING:
388 assert isinstance(entity, _StrPathToken)
389 return TokenRegistry(self, PathToken._intern[entity])
390 else:
391 try:
392 return entity._path_registry # type: ignore
393 except AttributeError:
394 raise IndexError(
395 f"invalid argument for RootRegistry.__getitem__: {entity}"
396 )
397
398 def _truncate_recursive(self) -> RootRegistry:
399 return self
400
401 if not TYPE_CHECKING:
402 __getitem__ = _getitem
403
404
405PathRegistry.root = RootRegistry()
406
407
408class PathToken(orm_base.InspectionAttr, HasCacheKey, str):
409 """cacheable string token"""
410
411 _intern: Dict[str, PathToken] = {}
412
413 def _gen_cache_key(
414 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
415 ) -> Tuple[Any, ...]:
416 return (str(self),)
417
418 @property
419 def _path_for_compare(self) -> Optional[_PathRepresentation]:
420 return None
421
422 @classmethod
423 def intern(cls, strvalue: str) -> PathToken:
424 if strvalue in cls._intern:
425 return cls._intern[strvalue]
426 else:
427 cls._intern[strvalue] = result = PathToken(strvalue)
428 return result
429
430
431class TokenRegistry(PathRegistry):
432 __slots__ = ("token", "parent", "path", "natural_path")
433
434 inherit_cache = True
435
436 token: _StrPathToken
437 parent: CreatesToken
438
439 def __init__(self, parent: CreatesToken, token: _StrPathToken):
440 token = PathToken.intern(token)
441
442 self.token = token
443 self.parent = parent
444 self.path = parent.path + (token,)
445 self.natural_path = parent.natural_path + (token,)
446
447 has_entity = False
448
449 is_token = True
450
451 def generate_for_superclasses(self) -> Iterator[PathRegistry]:
452 # NOTE: this method is no longer used. consider removal
453 parent = self.parent
454 if is_root(parent):
455 yield self
456 return
457
458 if TYPE_CHECKING:
459 assert isinstance(parent, AbstractEntityRegistry)
460 if not parent.is_aliased_class:
461 for mp_ent in parent.mapper.iterate_to_root():
462 yield TokenRegistry(parent.parent[mp_ent], self.token)
463 elif (
464 parent.is_aliased_class
465 and cast(
466 "AliasedInsp[Any]",
467 parent.entity,
468 )._is_with_polymorphic
469 ):
470 yield self
471 for ent in cast(
472 "AliasedInsp[Any]", parent.entity
473 )._with_polymorphic_entities:
474 yield TokenRegistry(parent.parent[ent], self.token)
475 else:
476 yield self
477
478 def _generate_natural_for_superclasses(
479 self,
480 ) -> Iterator[_PathRepresentation]:
481 parent = self.parent
482 if is_root(parent):
483 yield self.natural_path
484 return
485
486 if TYPE_CHECKING:
487 assert isinstance(parent, AbstractEntityRegistry)
488 for mp_ent in parent.mapper.iterate_to_root():
489 yield TokenRegistry(parent.parent[mp_ent], self.token).natural_path
490 if (
491 parent.is_aliased_class
492 and cast(
493 "AliasedInsp[Any]",
494 parent.entity,
495 )._is_with_polymorphic
496 ):
497 yield self.natural_path
498 for ent in cast(
499 "AliasedInsp[Any]", parent.entity
500 )._with_polymorphic_entities:
501 yield (
502 TokenRegistry(parent.parent[ent], self.token).natural_path
503 )
504 else:
505 yield self.natural_path
506
507 def _getitem(self, entity: Any) -> Any:
508 try:
509 return self.path[entity]
510 except TypeError as err:
511 raise IndexError(f"{entity}") from err
512
513 if not TYPE_CHECKING:
514 __getitem__ = _getitem
515
516
517class PropRegistry(PathRegistry):
518 __slots__ = (
519 "prop",
520 "parent",
521 "path",
522 "natural_path",
523 "has_entity",
524 "entity",
525 "mapper",
526 "_wildcard_path_loader_key",
527 "_default_path_loader_key",
528 "_loader_key",
529 "is_unnatural",
530 )
531 inherit_cache = True
532 is_property = True
533
534 prop: StrategizedProperty[Any]
535 mapper: Optional[Mapper[Any]]
536 entity: Optional[_InternalEntityType[Any]]
537
538 def __init__(
539 self, parent: AbstractEntityRegistry, prop: StrategizedProperty[Any]
540 ):
541
542 # restate this path in terms of the
543 # given StrategizedProperty's parent.
544 insp = cast("_InternalEntityType[Any]", parent[-1])
545 natural_parent: AbstractEntityRegistry = parent
546
547 # inherit "is_unnatural" from the parent
548 self.is_unnatural = parent.parent.is_unnatural or bool(
549 parent.mapper.inherits
550 )
551
552 if not insp.is_aliased_class or insp._use_mapper_path: # type: ignore
553 parent = natural_parent = parent.parent[prop.parent]
554 elif (
555 insp.is_aliased_class
556 and insp.with_polymorphic_mappers
557 and prop.parent in insp.with_polymorphic_mappers
558 ):
559 subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent) # type: ignore # noqa: E501
560 parent = parent.parent[subclass_entity]
561
562 # when building a path where with_polymorphic() is in use,
563 # special logic to determine the "natural path" when subclass
564 # entities are used.
565 #
566 # here we are trying to distinguish between a path that starts
567 # on a with_polymorphic entity vs. one that starts on a
568 # normal entity that introduces a with_polymorphic() in the
569 # middle using of_type():
570 #
571 # # as in test_polymorphic_rel->
572 # # test_subqueryload_on_subclass_uses_path_correctly
573 # wp = with_polymorphic(RegularEntity, "*")
574 # sess.query(wp).options(someload(wp.SomeSubEntity.foos))
575 #
576 # vs
577 #
578 # # as in test_relationship->JoinedloadWPolyOfTypeContinued
579 # wp = with_polymorphic(SomeFoo, "*")
580 # sess.query(RegularEntity).options(
581 # someload(RegularEntity.foos.of_type(wp))
582 # .someload(wp.SubFoo.bar)
583 # )
584 #
585 # in the former case, the Query as it generates a path that we
586 # want to match will be in terms of the with_polymorphic at the
587 # beginning. in the latter case, Query will generate simple
588 # paths that don't know about this with_polymorphic, so we must
589 # use a separate natural path.
590 #
591 #
592 if parent.parent:
593 natural_parent = parent.parent[subclass_entity.mapper]
594 self.is_unnatural = True
595 else:
596 natural_parent = parent
597 elif (
598 natural_parent.parent
599 and insp.is_aliased_class
600 and prop.parent # this should always be the case here
601 is not insp.mapper
602 and insp.mapper.isa(prop.parent)
603 ):
604 natural_parent = parent.parent[prop.parent]
605
606 self.prop = prop
607 self.parent = parent
608 self.path = parent.path + (prop,)
609 self.natural_path = natural_parent.natural_path + (prop,)
610
611 self.has_entity = prop._links_to_entity
612 if prop._is_relationship:
613 if TYPE_CHECKING:
614 assert isinstance(prop, RelationshipProperty)
615 self.entity = prop.entity
616 self.mapper = prop.mapper
617 else:
618 self.entity = None
619 self.mapper = None
620
621 self._wildcard_path_loader_key = (
622 "loader",
623 parent.natural_path + self.prop._wildcard_token,
624 )
625 self._default_path_loader_key = self.prop._default_path_loader_key
626 self._loader_key = ("loader", self.natural_path)
627
628 def _truncate_recursive(self) -> PropRegistry:
629 earliest = None
630 for i, token in enumerate(reversed(self.path[:-1])):
631 if token is self.prop:
632 earliest = i
633
634 if earliest is None:
635 return self
636 else:
637 return self.coerce(self.path[0 : -(earliest + 1)]) # type: ignore
638
639 @property
640 def entity_path(self) -> AbstractEntityRegistry:
641 assert self.entity is not None
642 return self[self.entity]
643
644 def _getitem(
645 self, entity: Union[int, slice, _InternalEntityType[Any]]
646 ) -> Union[AbstractEntityRegistry, _PathElementType, _PathRepresentation]:
647 if isinstance(entity, (int, slice)):
648 return self.path[entity]
649 else:
650 return SlotsEntityRegistry(self, entity)
651
652 if not TYPE_CHECKING:
653 __getitem__ = _getitem
654
655
656class AbstractEntityRegistry(CreatesToken):
657 __slots__ = (
658 "key",
659 "parent",
660 "is_aliased_class",
661 "path",
662 "entity",
663 "natural_path",
664 )
665
666 has_entity = True
667 is_entity = True
668
669 parent: Union[RootRegistry, PropRegistry]
670 key: _InternalEntityType[Any]
671 entity: _InternalEntityType[Any]
672 is_aliased_class: bool
673
674 def __init__(
675 self,
676 parent: Union[RootRegistry, PropRegistry],
677 entity: _InternalEntityType[Any],
678 ):
679 self.key = entity
680 self.parent = parent
681 self.is_aliased_class = entity.is_aliased_class
682 self.entity = entity
683 self.path = parent.path + (entity,)
684
685 # the "natural path" is the path that we get when Query is traversing
686 # from the lead entities into the various relationships; it corresponds
687 # to the structure of mappers and relationships. when we are given a
688 # path that comes from loader options, as of 1.3 it can have ac-hoc
689 # with_polymorphic() and other AliasedInsp objects inside of it, which
690 # are usually not present in mappings. So here we track both the
691 # "enhanced" path in self.path and the "natural" path that doesn't
692 # include those objects so these two traversals can be matched up.
693
694 # the test here for "(self.is_aliased_class or parent.is_unnatural)"
695 # are to avoid the more expensive conditional logic that follows if we
696 # know we don't have to do it. This conditional can just as well be
697 # "if parent.path:", it just is more function calls.
698 #
699 # This is basically the only place that the "is_unnatural" flag
700 # actually changes behavior.
701 if parent.path and (self.is_aliased_class or parent.is_unnatural):
702 # this is an infrequent code path used for loader strategies that
703 # also make use of of_type() or other intricate polymorphic
704 # base/subclass combinations
705 parent_natural_entity = parent.natural_path[-1]
706
707 if entity.mapper.isa(
708 parent_natural_entity.mapper # type: ignore
709 ) or parent_natural_entity.mapper.isa( # type: ignore
710 entity.mapper
711 ):
712 # when the entity mapper and parent mapper are in an
713 # inheritance relationship, use entity.mapper in natural_path.
714 # First case: entity.mapper inherits from parent mapper (e.g.,
715 # accessing a subclass mapper through parent path). Second case
716 # (issue #13193): parent mapper inherits from entity.mapper
717 # (e.g., parent path has Sub(Base) but we're accessing with
718 # Base where Base.related is declared, so use Base in
719 # natural_path).
720 self.natural_path = parent.natural_path + (entity.mapper,)
721 else:
722 self.natural_path = parent.natural_path + (
723 parent_natural_entity.entity, # type: ignore
724 )
725 # it seems to make sense that since these paths get mixed up
726 # with statements that are cached or not, we should make
727 # sure the natural path is cacheable across different occurrences
728 # of equivalent AliasedClass objects. however, so far this
729 # does not seem to be needed for whatever reason.
730 # elif not parent.path and self.is_aliased_class:
731 # self.natural_path = (self.entity._generate_cache_key()[0], )
732 else:
733 self.natural_path = self.path
734
735 def _truncate_recursive(self) -> AbstractEntityRegistry:
736 return self.parent._truncate_recursive()[self.entity]
737
738 @property
739 def root_entity(self) -> _InternalEntityType[Any]:
740 return self.odd_element(0)
741
742 @property
743 def entity_path(self) -> PathRegistry:
744 return self
745
746 @property
747 def mapper(self) -> Mapper[Any]:
748 return self.entity.mapper
749
750 def __bool__(self) -> bool:
751 return True
752
753 def _getitem(
754 self, entity: Any
755 ) -> Union[_PathElementType, _PathRepresentation, PathRegistry]:
756 if isinstance(entity, (int, slice)):
757 return self.path[entity]
758 elif entity in PathToken._intern:
759 return TokenRegistry(self, PathToken._intern[entity])
760 else:
761 return PropRegistry(self, entity)
762
763 if not TYPE_CHECKING:
764 __getitem__ = _getitem
765
766
767class SlotsEntityRegistry(AbstractEntityRegistry):
768 # for aliased class, return lightweight, no-cycles created
769 # version
770 inherit_cache = True
771
772
773class _ERDict(Dict[Any, Any]):
774 def __init__(self, registry: CachingEntityRegistry):
775 self.registry = registry
776
777 def __missing__(self, key: Any) -> PropRegistry:
778 self[key] = item = PropRegistry(self.registry, key)
779
780 return item
781
782
783class CachingEntityRegistry(AbstractEntityRegistry):
784 # for long lived mapper, return dict based caching
785 # version that creates reference cycles
786
787 __slots__ = ("_cache",)
788
789 inherit_cache = True
790
791 def __init__(
792 self,
793 parent: Union[RootRegistry, PropRegistry],
794 entity: _InternalEntityType[Any],
795 ):
796 super().__init__(parent, entity)
797 self._cache = _ERDict(self)
798
799 def pop(self, key: Any, default: Any) -> Any:
800 return self._cache.pop(key, default)
801
802 def _getitem(self, entity: Any) -> Any:
803 if isinstance(entity, (int, slice)):
804 return self.path[entity]
805 elif isinstance(entity, PathToken):
806 return TokenRegistry(self, entity)
807 else:
808 return self._cache[entity]
809
810 if not TYPE_CHECKING:
811 __getitem__ = _getitem
812
813
814if TYPE_CHECKING:
815
816 def path_is_entity(
817 path: PathRegistry,
818 ) -> TypeGuard[AbstractEntityRegistry]: ...
819
820 def path_is_property(path: PathRegistry) -> TypeGuard[PropRegistry]: ...
821
822else:
823 path_is_entity = operator.attrgetter("is_entity")
824 path_is_property = operator.attrgetter("is_property")
825