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

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

412 statements  

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 inspection 

31from .. import util 

32from ..sql import visitors 

33from ..sql.cache_key import HasCacheKey 

34 

35if TYPE_CHECKING: 

36 from typing import TypeGuard 

37 

38 from ._typing import _InternalEntityType 

39 from .interfaces import StrategizedProperty 

40 from .mapper import Mapper 

41 from .relationships import RelationshipProperty 

42 from .util import AliasedInsp 

43 from ..sql.cache_key import _CacheKeyTraversalType 

44 from ..sql.elements import BindParameter 

45 from ..sql.visitors import anon_map 

46 from ..util.typing import _LiteralStar 

47 

48 def is_root(path: PathRegistry) -> TypeGuard[RootRegistry]: ... 

49 

50 def is_entity( 

51 path: PathRegistry, 

52 ) -> TypeGuard[_AbstractEntityRegistry]: ... 

53 

54else: 

55 is_root = operator.attrgetter("is_root") 

56 is_entity = operator.attrgetter("is_entity") 

57 

58 

59_SerializedPath = List[Any] 

60_StrPathToken = str 

61_PathElementType = Union[ 

62 _StrPathToken, "_InternalEntityType[Any]", "StrategizedProperty[Any]" 

63] 

64 

65# the representation is in fact 

66# a tuple with alternating: 

67# [_InternalEntityType[Any], Union[str, StrategizedProperty[Any]], 

68# _InternalEntityType[Any], Union[str, StrategizedProperty[Any]], ...] 

69# this might someday be a tuple of 2-tuples instead, but paths can be 

70# chopped at odd intervals as well so this is less flexible 

71_PathRepresentation = Tuple[_PathElementType, ...] 

72 

73# NOTE: these names are weird since the array is 0-indexed, 

74# the "_Odd" entries are at 0, 2, 4, etc 

75_OddPathRepresentation = Sequence["_InternalEntityType[Any]"] 

76_EvenPathRepresentation = Sequence[Union["StrategizedProperty[Any]", str]] 

77 

78 

79log = logging.getLogger(__name__) 

80 

81 

82def _unreduce_path(path: _SerializedPath) -> PathRegistry: 

83 return PathRegistry.deserialize(path) 

84 

85 

86_WILDCARD_TOKEN: _LiteralStar = "*" 

87_DEFAULT_TOKEN = "_sa_default" 

88 

89 

90@inspection._self_inspects 

91class PathRegistry(HasCacheKey): 

92 """Represent query load paths and registry functions. 

93 

94 Basically represents structures like: 

95 

96 (<User mapper>, "orders", <Order mapper>, "items", <Item mapper>) 

97 

98 These structures are generated by things like 

99 query options (joinedload(), subqueryload(), etc.) and are 

100 used to compose keys stored in the query._attributes dictionary 

101 for various options. 

102 

103 They are then re-composed at query compile/result row time as 

104 the query is formed and as rows are fetched, where they again 

105 serve to compose keys to look up options in the context.attributes 

106 dictionary, which is copied from query._attributes. 

107 

108 The path structure has a limited amount of caching, where each 

109 "root" ultimately pulls from a fixed registry associated with 

110 the first mapper, that also contains elements for each of its 

111 property keys. However paths longer than two elements, which 

112 are the exception rather than the rule, are generated on an 

113 as-needed basis. 

114 

115 """ 

116 

117 __slots__ = () 

118 

119 is_token = False 

120 is_root = False 

121 has_entity = False 

122 is_property = False 

123 is_entity = False 

124 

125 is_unnatural: bool 

126 

127 path: _PathRepresentation 

128 natural_path: _PathRepresentation 

129 parent: Optional[PathRegistry] 

130 root: RootRegistry 

131 

132 _cache_key_traversal: _CacheKeyTraversalType = [ 

133 ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key_list) 

134 ] 

135 

136 def __eq__(self, other: Any) -> bool: 

137 try: 

138 return other is not None and self.path == other._path_for_compare 

139 except AttributeError: 

140 util.warn( 

141 "Comparison of PathRegistry to %r is not supported" 

142 % (type(other)) 

143 ) 

144 return False 

145 

146 def __ne__(self, other: Any) -> bool: 

147 try: 

148 return other is None or self.path != other._path_for_compare 

149 except AttributeError: 

150 util.warn( 

151 "Comparison of PathRegistry to %r is not supported" 

152 % (type(other)) 

153 ) 

154 return True 

155 

156 @property 

157 def _path_for_compare(self) -> Optional[_PathRepresentation]: 

158 return self.path 

159 

160 def odd_element(self, index: int) -> _InternalEntityType[Any]: 

161 return self.path[index] # type: ignore[return-value] 

162 

163 def set(self, attributes: Dict[Any, Any], key: Any, value: Any) -> None: 

164 log.debug("set '%s' on path '%s' to '%s'", key, self, value) 

165 attributes[(key, self.natural_path)] = value 

166 

167 def setdefault( 

168 self, attributes: Dict[Any, Any], key: Any, value: Any 

169 ) -> None: 

170 log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value) 

171 attributes.setdefault((key, self.natural_path), value) 

172 

173 def get( 

174 self, attributes: Dict[Any, Any], key: Any, value: Optional[Any] = None 

175 ) -> Any: 

176 key = (key, self.natural_path) 

177 if key in attributes: 

178 return attributes[key] 

179 else: 

180 return value 

181 

182 def __len__(self) -> int: 

183 return len(self.path) 

184 

185 def __hash__(self) -> int: 

186 return id(self) 

187 

188 @overload 

189 def __getitem__(self, entity: _StrPathToken) -> _TokenRegistry: ... 

190 

191 @overload 

192 def __getitem__(self, entity: int) -> _PathElementType: ... 

193 

194 @overload 

195 def __getitem__(self, entity: slice) -> _PathRepresentation: ... 

196 

197 @overload 

198 def __getitem__( 

199 self, entity: _InternalEntityType[Any] 

200 ) -> _AbstractEntityRegistry: ... 

201 

202 @overload 

203 def __getitem__( 

204 self, entity: StrategizedProperty[Any] 

205 ) -> _PropRegistry: ... 

206 

207 def __getitem__( 

208 self, 

209 entity: Union[ 

210 _StrPathToken, 

211 int, 

212 slice, 

213 _InternalEntityType[Any], 

214 StrategizedProperty[Any], 

215 ], 

216 ) -> Union[ 

217 _TokenRegistry, 

218 _PathElementType, 

219 _PathRepresentation, 

220 _PropRegistry, 

221 _AbstractEntityRegistry, 

222 ]: 

223 raise NotImplementedError() 

224 

225 # TODO: what are we using this for? 

226 @property 

227 def length(self) -> int: 

228 return len(self.path) 

229 

230 def pairs( 

231 self, 

232 ) -> Iterator[ 

233 Tuple[_InternalEntityType[Any], Union[str, StrategizedProperty[Any]]] 

234 ]: 

235 odd_path = cast(_OddPathRepresentation, self.path) 

236 even_path = cast(_EvenPathRepresentation, odd_path) 

237 for i in range(0, len(odd_path), 2): 

238 yield odd_path[i], even_path[i + 1] 

239 

240 def contains_mapper(self, mapper: Mapper[Any]) -> bool: 

241 _m_path = cast(_OddPathRepresentation, self.path) 

242 for path_mapper in [_m_path[i] for i in range(0, len(_m_path), 2)]: 

243 if path_mapper.mapper.isa(mapper): 

244 return True 

245 else: 

246 return False 

247 

248 def contains(self, attributes: Dict[Any, Any], key: Any) -> bool: 

249 return (key, self.path) in attributes 

250 

251 def __reduce__(self) -> Any: 

252 return _unreduce_path, (self.serialize(),) 

253 

254 @classmethod 

255 def _serialize_path(cls, path: _PathRepresentation) -> _SerializedPath: 

256 _m_path = cast(_OddPathRepresentation, path) 

257 _p_path = cast(_EvenPathRepresentation, path) 

258 

259 return list( 

260 zip( 

261 tuple( 

262 m.class_ if (m.is_mapper or m.is_aliased_class) else str(m) 

263 for m in [_m_path[i] for i in range(0, len(_m_path), 2)] 

264 ), 

265 tuple( 

266 p.key if insp_is_mapper_property(p) else str(p) 

267 for p in [_p_path[i] for i in range(1, len(_p_path), 2)] 

268 ) 

269 + (None,), 

270 ) 

271 ) 

272 

273 @classmethod 

274 def _deserialize_path(cls, path: _SerializedPath) -> _PathRepresentation: 

275 def _deserialize_mapper_token(mcls: Any) -> Any: 

276 return ( 

277 # note: we likely dont want configure=True here however 

278 # this is maintained at the moment for backwards compatibility 

279 orm_base._inspect_mapped_class(mcls, configure=True) 

280 if mcls not in PathToken._intern 

281 else PathToken._intern[mcls] 

282 ) 

283 

284 def _deserialize_key_token(mcls: Any, key: Any) -> Any: 

285 if key is None: 

286 return None 

287 elif key in PathToken._intern: 

288 return PathToken._intern[key] 

289 else: 

290 mp = orm_base._inspect_mapped_class(mcls, configure=True) 

291 assert mp is not None 

292 return mp.attrs[key] 

293 

294 p = tuple( 

295 chain( 

296 *[ 

297 ( 

298 _deserialize_mapper_token(mcls), 

299 _deserialize_key_token(mcls, key), 

300 ) 

301 for mcls, key in path 

302 ] 

303 ) 

304 ) 

305 if p and p[-1] is None: 

306 p = p[0:-1] 

307 return p 

308 

309 def serialize(self) -> _SerializedPath: 

310 path = self.path 

311 return self._serialize_path(path) 

312 

313 @classmethod 

314 def deserialize(cls, path: _SerializedPath) -> PathRegistry: 

315 assert path is not None 

316 p = cls._deserialize_path(path) 

317 return cls.coerce(p) 

318 

319 @overload 

320 @classmethod 

321 def per_mapper(cls, mapper: Mapper[Any]) -> _CachingEntityRegistry: ... 

322 

323 @overload 

324 @classmethod 

325 def per_mapper(cls, mapper: AliasedInsp[Any]) -> _SlotsEntityRegistry: ... 

326 

327 @classmethod 

328 def per_mapper( 

329 cls, mapper: _InternalEntityType[Any] 

330 ) -> _AbstractEntityRegistry: 

331 if mapper.is_mapper: 

332 return _CachingEntityRegistry(cls.root, mapper) 

333 else: 

334 return _SlotsEntityRegistry(cls.root, mapper) 

335 

336 @classmethod 

337 def coerce(cls, raw: _PathRepresentation) -> PathRegistry: 

338 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry: 

339 return prev[next_] 

340 

341 # can't quite get mypy to appreciate this one :) 

342 return reduce(_red, raw, cls.root) # type: ignore[arg-type] 

343 

344 def __add__(self, other: PathRegistry) -> PathRegistry: 

345 def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry: 

346 return prev[next_] 

347 

348 return reduce(_red, other.path, self) 

349 

350 def __str__(self) -> str: 

351 return f"ORM Path[{' -> '.join(str(elem) for elem in self.path)}]" 

352 

353 def __repr__(self) -> str: 

354 return f"{self.__class__.__name__}({self.path!r})" 

355 

356 def path_string(self) -> str: 

357 """Return a user-facing string representation of this path, 

358 e.g. ``"User.orders -> Order.items"``. 

359 

360 """ 

361 

362 raw = self.path 

363 parts = [] 

364 lraw = len(raw) 

365 for i in range(0, lraw - 1, 2): 

366 entity = raw[i] 

367 prop = raw[i + 1] 

368 prop_key = getattr(prop, "key", str(prop)) 

369 

370 if ( 

371 i < lraw - 2 

372 and cast( 

373 "_InternalEntityType[Any]", raw[i + 2] 

374 ).is_aliased_class 

375 ): 

376 parts.append( 

377 f"{orm_base.entity_str(entity)}.{prop_key}." 

378 f"of_type({orm_base.entity_str(raw[i + 2])})" 

379 ) 

380 else: 

381 parts.append(f"{orm_base.entity_str(entity)}.{prop_key}") 

382 

383 return ( 

384 " -> ".join(parts) 

385 if parts 

386 else orm_base.entity_str(self.path[0]) if self.path else "" 

387 ) 

388 

389 

390class _CreatesToken(PathRegistry): 

391 __slots__ = () 

392 

393 is_aliased_class: bool 

394 is_root: bool 

395 

396 def token(self, token: _StrPathToken) -> _TokenRegistry: 

397 if token.endswith(f":{_WILDCARD_TOKEN}"): 

398 return _TokenRegistry(self, token) 

399 elif token.endswith(f":{_DEFAULT_TOKEN}"): 

400 return _TokenRegistry(self.root, token) 

401 else: 

402 raise exc.ArgumentError(f"invalid token: {token}") 

403 

404 

405class RootRegistry(_CreatesToken): 

406 """Root registry, defers to mappers so that 

407 paths are maintained per-root-mapper. 

408 

409 """ 

410 

411 __slots__ = () 

412 

413 inherit_cache = True 

414 

415 path = natural_path = () 

416 has_entity = False 

417 is_aliased_class = False 

418 is_root = True 

419 is_unnatural = False 

420 

421 def _getitem( 

422 self, entity: Any 

423 ) -> Union[_TokenRegistry, _AbstractEntityRegistry]: 

424 if entity in PathToken._intern: 

425 if TYPE_CHECKING: 

426 assert isinstance(entity, _StrPathToken) 

427 return _TokenRegistry(self, PathToken._intern[entity]) 

428 else: 

429 try: 

430 return entity._path_registry # type: ignore[no-any-return] 

431 except AttributeError: 

432 raise IndexError( 

433 f"invalid argument for RootRegistry.__getitem__: {entity}" 

434 ) 

435 

436 def _truncate_recursive(self) -> RootRegistry: 

437 return self 

438 

439 if not TYPE_CHECKING: 

440 __getitem__ = _getitem 

441 

442 

443PathRegistry.root = RootRegistry() 

444 

445 

446class PathToken(orm_base.InspectionAttr, HasCacheKey, str): 

447 """cacheable string token""" 

448 

449 _intern: Dict[str, PathToken] = {} 

450 

451 def _gen_cache_key( 

452 self, anon_map: anon_map, bindparams: List[BindParameter[Any]] 

453 ) -> Tuple[Any, ...]: 

454 return (str(self),) 

455 

456 @property 

457 def _path_for_compare(self) -> Optional[_PathRepresentation]: 

458 return None 

459 

460 @classmethod 

461 def intern(cls, strvalue: str) -> PathToken: 

462 if strvalue in cls._intern: 

463 return cls._intern[strvalue] 

464 else: 

465 cls._intern[strvalue] = result = PathToken(strvalue) 

466 return result 

467 

468 

469class _TokenRegistry(PathRegistry): 

470 __slots__ = ("token", "parent", "path", "natural_path") 

471 

472 inherit_cache = True 

473 

474 token: _StrPathToken 

475 parent: _CreatesToken 

476 

477 def __init__(self, parent: _CreatesToken, token: _StrPathToken): 

478 token = PathToken.intern(token) 

479 

480 self.token = token 

481 self.parent = parent 

482 self.path = parent.path + (token,) 

483 self.natural_path = parent.natural_path + (token,) 

484 

485 has_entity = False 

486 

487 is_token = True 

488 

489 def generate_for_superclasses(self) -> Iterator[PathRegistry]: 

490 # NOTE: this method is no longer used. consider removal 

491 parent = self.parent 

492 if is_root(parent): 

493 yield self 

494 return 

495 

496 if TYPE_CHECKING: 

497 assert isinstance(parent, _AbstractEntityRegistry) 

498 if not parent.is_aliased_class: 

499 for mp_ent in parent.mapper.iterate_to_root(): 

500 yield _TokenRegistry(parent.parent[mp_ent], self.token) 

501 elif ( 

502 parent.is_aliased_class 

503 and cast( 

504 "AliasedInsp[Any]", 

505 parent.entity, 

506 )._is_with_polymorphic 

507 ): 

508 yield self 

509 for ent in cast( 

510 "AliasedInsp[Any]", parent.entity 

511 )._with_polymorphic_entities: 

512 yield _TokenRegistry(parent.parent[ent], self.token) 

513 else: 

514 yield self 

515 

516 def _generate_natural_for_superclasses( 

517 self, 

518 ) -> Iterator[_PathRepresentation]: 

519 parent = self.parent 

520 if is_root(parent): 

521 yield self.natural_path 

522 return 

523 

524 if TYPE_CHECKING: 

525 assert isinstance(parent, _AbstractEntityRegistry) 

526 for mp_ent in parent.mapper.iterate_to_root(): 

527 yield _TokenRegistry( 

528 parent.parent[mp_ent], self.token 

529 ).natural_path 

530 if ( 

531 parent.is_aliased_class 

532 and cast( 

533 "AliasedInsp[Any]", 

534 parent.entity, 

535 )._is_with_polymorphic 

536 ): 

537 yield self.natural_path 

538 for ent in cast( 

539 "AliasedInsp[Any]", parent.entity 

540 )._with_polymorphic_entities: 

541 yield ( 

542 _TokenRegistry(parent.parent[ent], self.token).natural_path 

543 ) 

544 else: 

545 yield self.natural_path 

546 

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

548 try: 

549 return self.path[entity] 

550 except TypeError as err: 

551 raise IndexError(f"{entity}") from err 

552 

553 if not TYPE_CHECKING: 

554 __getitem__ = _getitem 

555 

556 

557class _PropRegistry(PathRegistry): 

558 __slots__ = ( 

559 "prop", 

560 "parent", 

561 "path", 

562 "natural_path", 

563 "has_entity", 

564 "entity", 

565 "mapper", 

566 "_wildcard_path_loader_key", 

567 "_default_path_loader_key", 

568 "_loader_key", 

569 "is_unnatural", 

570 ) 

571 inherit_cache = True 

572 is_property = True 

573 

574 prop: StrategizedProperty[Any] 

575 mapper: Optional[Mapper[Any]] 

576 entity: Optional[_InternalEntityType[Any]] 

577 parent: _AbstractEntityRegistry 

578 

579 def __init__( 

580 self, parent: _AbstractEntityRegistry, prop: StrategizedProperty[Any] 

581 ): 

582 

583 # restate this path in terms of the 

584 # given StrategizedProperty's parent. 

585 insp = cast("_InternalEntityType[Any]", parent[-1]) 

586 natural_parent: _AbstractEntityRegistry = parent 

587 

588 # inherit "is_unnatural" from the parent 

589 self.is_unnatural = parent.parent.is_unnatural or bool( 

590 parent.mapper.inherits 

591 ) 

592 

593 if not insp.is_aliased_class or insp._use_mapper_path: # type: ignore[union-attr] # noqa: E501 

594 parent = natural_parent = parent.parent[prop.parent] 

595 elif ( 

596 insp.is_aliased_class 

597 and insp.with_polymorphic_mappers 

598 and prop.parent in insp.with_polymorphic_mappers 

599 ): 

600 subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent) # type: ignore[union-attr] # noqa: E501 

601 parent = parent.parent[subclass_entity] 

602 

603 # when building a path where with_polymorphic() is in use, 

604 # special logic to determine the "natural path" when subclass 

605 # entities are used. 

606 # 

607 # here we are trying to distinguish between a path that starts 

608 # on a with_polymorphic entity vs. one that starts on a 

609 # normal entity that introduces a with_polymorphic() in the 

610 # middle using of_type(): 

611 # 

612 # # as in test_polymorphic_rel-> 

613 # # test_subqueryload_on_subclass_uses_path_correctly 

614 # wp = with_polymorphic(RegularEntity, "*") 

615 # sess.query(wp).options(someload(wp.SomeSubEntity.foos)) 

616 # 

617 # vs 

618 # 

619 # # as in test_relationship->JoinedloadWPolyOfTypeContinued 

620 # wp = with_polymorphic(SomeFoo, "*") 

621 # sess.query(RegularEntity).options( 

622 # someload(RegularEntity.foos.of_type(wp)) 

623 # .someload(wp.SubFoo.bar) 

624 # ) 

625 # 

626 # in the former case, the Query as it generates a path that we 

627 # want to match will be in terms of the with_polymorphic at the 

628 # beginning. in the latter case, Query will generate simple 

629 # paths that don't know about this with_polymorphic, so we must 

630 # use a separate natural path. 

631 # 

632 # 

633 if parent.parent: 

634 natural_parent = parent.parent[subclass_entity.mapper] 

635 self.is_unnatural = True 

636 else: 

637 natural_parent = parent 

638 elif ( 

639 natural_parent.parent 

640 and insp.is_aliased_class 

641 and prop.parent # this should always be the case here 

642 is not insp.mapper 

643 and insp.mapper.isa(prop.parent) 

644 ): 

645 natural_parent = parent.parent[prop.parent] 

646 

647 self.prop = prop 

648 self.parent = parent 

649 self.path = parent.path + (prop,) 

650 self.natural_path = natural_parent.natural_path + (prop,) 

651 

652 self.has_entity = prop._links_to_entity 

653 if prop._is_relationship: 

654 if TYPE_CHECKING: 

655 assert isinstance(prop, RelationshipProperty) 

656 self.entity = prop.entity 

657 self.mapper = prop.mapper 

658 else: 

659 self.entity = None 

660 self.mapper = None 

661 

662 self._wildcard_path_loader_key = ( 

663 "loader", 

664 parent.natural_path + self.prop._wildcard_token, 

665 ) 

666 self._default_path_loader_key = self.prop._default_path_loader_key 

667 self._loader_key = ("loader", self.natural_path) 

668 

669 def _truncate_recursive(self) -> _PropRegistry: 

670 earliest = None 

671 for i, token in enumerate(reversed(self.path[:-1])): 

672 if token is self.prop: 

673 earliest = i 

674 

675 if earliest is None: 

676 return self 

677 else: 

678 return self.coerce(self.path[0 : -(earliest + 1)]) # type: ignore[return-value] # noqa: E501 

679 

680 @property 

681 def entity_path(self) -> _AbstractEntityRegistry: 

682 assert self.entity is not None 

683 return self[self.entity] 

684 

685 def _getitem( 

686 self, entity: Union[int, slice, _InternalEntityType[Any]] 

687 ) -> Union[_AbstractEntityRegistry, _PathElementType, _PathRepresentation]: 

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

689 return self.path[entity] 

690 else: 

691 return _SlotsEntityRegistry(self, entity) 

692 

693 if not TYPE_CHECKING: 

694 __getitem__ = _getitem 

695 

696 

697class _AbstractEntityRegistry(_CreatesToken): 

698 __slots__ = ( 

699 "key", 

700 "parent", 

701 "is_aliased_class", 

702 "path", 

703 "entity", 

704 "natural_path", 

705 ) 

706 

707 has_entity = True 

708 is_entity = True 

709 

710 parent: Union[RootRegistry, _PropRegistry] 

711 key: _InternalEntityType[Any] 

712 entity: _InternalEntityType[Any] 

713 is_aliased_class: bool 

714 

715 def __init__( 

716 self, 

717 parent: Union[RootRegistry, _PropRegistry], 

718 entity: _InternalEntityType[Any], 

719 ): 

720 self.key = entity 

721 self.parent = parent 

722 self.is_aliased_class = entity.is_aliased_class 

723 self.entity = entity 

724 self.path = parent.path + (entity,) 

725 

726 # the "natural path" is the path that we get when Query is traversing 

727 # from the lead entities into the various relationships; it corresponds 

728 # to the structure of mappers and relationships. when we are given a 

729 # path that comes from loader options, as of 1.3 it can have ac-hoc 

730 # with_polymorphic() and other AliasedInsp objects inside of it, which 

731 # are usually not present in mappings. So here we track both the 

732 # "enhanced" path in self.path and the "natural" path that doesn't 

733 # include those objects so these two traversals can be matched up. 

734 

735 # the test here for "(self.is_aliased_class or parent.is_unnatural)" 

736 # are to avoid the more expensive conditional logic that follows if we 

737 # know we don't have to do it. This conditional can just as well be 

738 # "if parent.path:", it just is more function calls. 

739 # 

740 # This is basically the only place that the "is_unnatural" flag 

741 # actually changes behavior. 

742 if parent.path and (self.is_aliased_class or parent.is_unnatural): 

743 # this is an infrequent code path used for loader strategies that 

744 # also make use of of_type() or other intricate polymorphic 

745 # base/subclass combinations 

746 parent_natural_entity = parent.natural_path[-1] 

747 

748 if entity.mapper.isa( 

749 parent_natural_entity.mapper # type: ignore[union-attr] 

750 ) or parent_natural_entity.mapper.isa( # type: ignore[union-attr] 

751 entity.mapper 

752 ): 

753 # when the entity mapper and parent mapper are in an 

754 # inheritance relationship, use entity.mapper in natural_path. 

755 # First case: entity.mapper inherits from parent mapper (e.g., 

756 # accessing a subclass mapper through parent path). Second case 

757 # (issue #13193): parent mapper inherits from entity.mapper 

758 # (e.g., parent path has Sub(Base) but we're accessing with 

759 # Base where Base.related is declared, so use Base in 

760 # natural_path). 

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

762 else: 

763 self.natural_path = parent.natural_path + ( 

764 parent_natural_entity.entity, # type: ignore[operator, union-attr] # noqa: E501 

765 ) 

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

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

768 # sure the natural path is cacheable across different occurrences 

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

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

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

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

773 else: 

774 self.natural_path = self.path 

775 

776 def _truncate_recursive(self) -> _AbstractEntityRegistry: 

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

778 

779 @property 

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

781 return self.odd_element(0) 

782 

783 @property 

784 def entity_path(self) -> PathRegistry: 

785 return self 

786 

787 @property 

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

789 return self.entity.mapper 

790 

791 def __bool__(self) -> bool: 

792 return True 

793 

794 def _getitem( 

795 self, entity: Any 

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

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

798 return self.path[entity] 

799 elif entity in PathToken._intern: 

800 return _TokenRegistry(self, PathToken._intern[entity]) 

801 else: 

802 return _PropRegistry(self, entity) 

803 

804 if not TYPE_CHECKING: 

805 __getitem__ = _getitem 

806 

807 

808class _SlotsEntityRegistry(_AbstractEntityRegistry): 

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

810 # version 

811 inherit_cache = True 

812 

813 

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

815 def __init__(self, registry: _CachingEntityRegistry): 

816 self.registry = registry 

817 

818 def __missing__(self, key: Any) -> _PropRegistry: 

819 self[key] = item = _PropRegistry(self.registry, key) 

820 

821 return item 

822 

823 

824class _CachingEntityRegistry(_AbstractEntityRegistry): 

825 # for long lived mapper, return dict based caching 

826 # version that creates reference cycles 

827 

828 __slots__ = ("_cache",) 

829 

830 inherit_cache = True 

831 

832 def __init__( 

833 self, 

834 parent: Union[RootRegistry, _PropRegistry], 

835 entity: _InternalEntityType[Any], 

836 ): 

837 super().__init__(parent, entity) 

838 self._cache = _ERDict(self) 

839 

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

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

842 

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

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

845 return self.path[entity] 

846 elif isinstance(entity, PathToken): 

847 return _TokenRegistry(self, entity) 

848 else: 

849 return self._cache[entity] 

850 

851 if not TYPE_CHECKING: 

852 __getitem__ = _getitem 

853 

854 

855if TYPE_CHECKING: 

856 

857 def path_is_entity( 

858 path: PathRegistry, 

859 ) -> TypeGuard[_AbstractEntityRegistry]: ... 

860 

861 def path_is_property(path: PathRegistry) -> TypeGuard[_PropRegistry]: ... 

862 

863else: 

864 path_is_entity = operator.attrgetter("is_entity") 

865 path_is_property = operator.attrgetter("is_property")