Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/path_registry.py: 58%

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

396 statements  

1# orm/path_registry.py 

2# Copyright (C) 2005-2025 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 only for loader strategies 

703 # that also make use of of_type(). 

704 if entity.mapper.isa(parent.natural_path[-1].mapper): # type: ignore # noqa: E501 

705 self.natural_path = parent.natural_path + (entity.mapper,) 

706 else: 

707 self.natural_path = parent.natural_path + ( 

708 parent.natural_path[-1].entity, # type: ignore 

709 ) 

710 # it seems to make sense that since these paths get mixed up 

711 # with statements that are cached or not, we should make 

712 # sure the natural path is cacheable across different occurrences 

713 # of equivalent AliasedClass objects. however, so far this 

714 # does not seem to be needed for whatever reason. 

715 # elif not parent.path and self.is_aliased_class: 

716 # self.natural_path = (self.entity._generate_cache_key()[0], ) 

717 else: 

718 self.natural_path = self.path 

719 

720 def _truncate_recursive(self) -> AbstractEntityRegistry: 

721 return self.parent._truncate_recursive()[self.entity] 

722 

723 @property 

724 def root_entity(self) -> _InternalEntityType[Any]: 

725 return self.odd_element(0) 

726 

727 @property 

728 def entity_path(self) -> PathRegistry: 

729 return self 

730 

731 @property 

732 def mapper(self) -> Mapper[Any]: 

733 return self.entity.mapper 

734 

735 def __bool__(self) -> bool: 

736 return True 

737 

738 def _getitem( 

739 self, entity: Any 

740 ) -> Union[_PathElementType, _PathRepresentation, PathRegistry]: 

741 if isinstance(entity, (int, slice)): 

742 return self.path[entity] 

743 elif entity in PathToken._intern: 

744 return TokenRegistry(self, PathToken._intern[entity]) 

745 else: 

746 return PropRegistry(self, entity) 

747 

748 if not TYPE_CHECKING: 

749 __getitem__ = _getitem 

750 

751 

752class SlotsEntityRegistry(AbstractEntityRegistry): 

753 # for aliased class, return lightweight, no-cycles created 

754 # version 

755 inherit_cache = True 

756 

757 

758class _ERDict(Dict[Any, Any]): 

759 def __init__(self, registry: CachingEntityRegistry): 

760 self.registry = registry 

761 

762 def __missing__(self, key: Any) -> PropRegistry: 

763 self[key] = item = PropRegistry(self.registry, key) 

764 

765 return item 

766 

767 

768class CachingEntityRegistry(AbstractEntityRegistry): 

769 # for long lived mapper, return dict based caching 

770 # version that creates reference cycles 

771 

772 __slots__ = ("_cache",) 

773 

774 inherit_cache = True 

775 

776 def __init__( 

777 self, 

778 parent: Union[RootRegistry, PropRegistry], 

779 entity: _InternalEntityType[Any], 

780 ): 

781 super().__init__(parent, entity) 

782 self._cache = _ERDict(self) 

783 

784 def pop(self, key: Any, default: Any) -> Any: 

785 return self._cache.pop(key, default) 

786 

787 def _getitem(self, entity: Any) -> Any: 

788 if isinstance(entity, (int, slice)): 

789 return self.path[entity] 

790 elif isinstance(entity, PathToken): 

791 return TokenRegistry(self, entity) 

792 else: 

793 return self._cache[entity] 

794 

795 if not TYPE_CHECKING: 

796 __getitem__ = _getitem 

797 

798 

799if TYPE_CHECKING: 

800 

801 def path_is_entity( 

802 path: PathRegistry, 

803 ) -> TypeGuard[AbstractEntityRegistry]: ... 

804 

805 def path_is_property(path: PathRegistry) -> TypeGuard[PropRegistry]: ... 

806 

807else: 

808 path_is_entity = operator.attrgetter("is_entity") 

809 path_is_property = operator.attrgetter("is_property")