Загрузить файлы в «venv/Lib/site-packages/sqlalchemy/orm»

venv/Lib/site-packages/sqlalchemy/orm/mapped_collection.py

@@ -0,0 +1,557 @@
+# orm/mapped_collection.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+
+from __future__ import annotations
+
+import operator
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Generic
+from typing import List
+from typing import Optional
+from typing import Sequence
+from typing import Tuple
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import TypeVar
+from typing import Union
+
+from . import base
+from .collections import collection
+from .collections import collection_adapter
+from .. import exc as sa_exc
+from .. import util
+from ..sql import coercions
+from ..sql import expression
+from ..sql import roles
+from ..util.langhelpers import Missing
+from ..util.langhelpers import MissingOr
+from ..util.typing import Literal
+
+if TYPE_CHECKING:
+    from . import AttributeEventToken
+    from . import Mapper
+    from .collections import CollectionAdapter
+    from ..sql.elements import ColumnElement
+
+_KT = TypeVar("_KT", bound=Any)
+_VT = TypeVar("_VT", bound=Any)
+
+
+class _PlainColumnGetter(Generic[_KT]):
+    """Plain column getter, stores collection of Column objects
+    directly.
+
+    Serializes to a :class:`._SerializableColumnGetterV2`
+    which has more expensive __call__() performance
+    and some rare caveats.
+
+    """
+
+    __slots__ = ("cols", "composite")
+
+    def __init__(self, cols: Sequence[ColumnElement[_KT]]) -> None:
+        self.cols = cols
+        self.composite = len(cols) > 1
+
+    def __reduce__(
+        self,
+    ) -> Tuple[
+        Type[_SerializableColumnGetterV2[_KT]],
+        Tuple[Sequence[Tuple[Optional[str], Optional[str]]]],
+    ]:
+        return _SerializableColumnGetterV2._reduce_from_cols(self.cols)
+
+    def _cols(self, mapper: Mapper[_KT]) -> Sequence[ColumnElement[_KT]]:
+        return self.cols
+
+    def __call__(self, value: _KT) -> MissingOr[Union[_KT, Tuple[_KT, ...]]]:
+        state = base.instance_state(value)
+        m = base._state_mapper(state)
+
+        key: List[_KT] = [
+            m._get_state_attr_by_column(state, state.dict, col)
+            for col in self._cols(m)
+        ]
+        if self.composite:
+            return tuple(key)
+        else:
+            obj = key[0]
+            if obj is None:
+                return Missing
+            else:
+                return obj
+
+
+class _SerializableColumnGetterV2(_PlainColumnGetter[_KT]):
+    """Updated serializable getter which deals with
+    multi-table mapped classes.
+
+    Two extremely unusual cases are not supported.
+    Mappings which have tables across multiple metadata
+    objects, or which are mapped to non-Table selectables
+    linked across inheriting mappers may fail to function
+    here.
+
+    """
+
+    __slots__ = ("colkeys",)
+
+    def __init__(
+        self, colkeys: Sequence[Tuple[Optional[str], Optional[str]]]
+    ) -> None:
+        self.colkeys = colkeys
+        self.composite = len(colkeys) > 1
+
+    def __reduce__(
+        self,
+    ) -> Tuple[
+        Type[_SerializableColumnGetterV2[_KT]],
+        Tuple[Sequence[Tuple[Optional[str], Optional[str]]]],
+    ]:
+        return self.__class__, (self.colkeys,)
+
+    @classmethod
+    def _reduce_from_cols(cls, cols: Sequence[ColumnElement[_KT]]) -> Tuple[
+        Type[_SerializableColumnGetterV2[_KT]],
+        Tuple[Sequence[Tuple[Optional[str], Optional[str]]]],
+    ]:
+        def _table_key(c: ColumnElement[_KT]) -> Optional[str]:
+            if not isinstance(c.table, expression.TableClause):
+                return None
+            else:
+                return c.table.key  # type: ignore
+
+        colkeys = [(c.key, _table_key(c)) for c in cols]
+        return _SerializableColumnGetterV2, (colkeys,)
+
+    def _cols(self, mapper: Mapper[_KT]) -> Sequence[ColumnElement[_KT]]:
+        cols: List[ColumnElement[_KT]] = []
+        metadata = getattr(mapper.local_table, "metadata", None)
+        for ckey, tkey in self.colkeys:
+            if tkey is None or metadata is None or tkey not in metadata:
+                cols.append(mapper.local_table.c[ckey])  # type: ignore
+            else:
+                cols.append(metadata.tables[tkey].c[ckey])
+        return cols
+
+
+def column_keyed_dict(
+    mapping_spec: Union[Type[_KT], Callable[[_KT], _VT]],
+    *,
+    ignore_unpopulated_attribute: bool = False,
+) -> Type[KeyFuncDict[_KT, _KT]]:
+    """A dictionary-based collection type with column-based keying.
+
+    .. versionchanged:: 2.0 Renamed :data:`.column_mapped_collection` to
+       :class:`.column_keyed_dict`.
+
+    Returns a :class:`.KeyFuncDict` factory which will produce new
+    dictionary keys based on the value of a particular :class:`.Column`-mapped
+    attribute on ORM mapped instances to be added to the dictionary.
+
+    .. note:: the value of the target attribute must be assigned with its
+       value at the time that the object is being added to the
+       dictionary collection.   Additionally, changes to the key attribute
+       are **not tracked**, which means the key in the dictionary is not
+       automatically synchronized with the key value on the target object
+       itself.  See :ref:`key_collections_mutations` for further details.
+
+    .. seealso::
+
+        :ref:`orm_dictionary_collection` - background on use
+
+    :param mapping_spec: a :class:`_schema.Column` object that is expected
+     to be mapped by the target mapper to a particular attribute on the
+     mapped class, the value of which on a particular instance is to be used
+     as the key for a new dictionary entry for that instance.
+    :param ignore_unpopulated_attribute:  if True, and the mapped attribute
+     indicated by the given :class:`_schema.Column` target attribute
+     on an object is not populated at all, the operation will be silently
+     skipped.  By default, an error is raised.
+
+     .. versionadded:: 2.0 an error is raised by default if the attribute
+        being used for the dictionary key is determined that it was never
+        populated with any value.  The
+        :paramref:`_orm.column_keyed_dict.ignore_unpopulated_attribute`
+        parameter may be set which will instead indicate that this condition
+        should be ignored, and the append operation silently skipped.
+        This is in contrast to the behavior of the 1.x series which would
+        erroneously populate the value in the dictionary with an arbitrary key
+        value of ``None``.
+
+
+    """
+    cols = [
+        coercions.expect(roles.ColumnArgumentRole, q, argname="mapping_spec")
+        for q in util.to_list(mapping_spec)
+    ]
+    keyfunc = _PlainColumnGetter(cols)
+    return _mapped_collection_cls(
+        keyfunc,
+        ignore_unpopulated_attribute=ignore_unpopulated_attribute,
+    )
+
+
+class _AttrGetter:
+    __slots__ = ("attr_name", "getter")
+
+    def __init__(self, attr_name: str):
+        self.attr_name = attr_name
+        self.getter = operator.attrgetter(attr_name)
+
+    def __call__(self, mapped_object: Any) -> Any:
+        obj = self.getter(mapped_object)
+        if obj is None:
+            state = base.instance_state(mapped_object)
+            mp = state.mapper
+            if self.attr_name in mp.attrs:
+                dict_ = state.dict
+                obj = dict_.get(self.attr_name, base.NO_VALUE)
+                if obj is None:
+                    return Missing
+            else:
+                return Missing
+
+        return obj
+
+    def __reduce__(self) -> Tuple[Type[_AttrGetter], Tuple[str]]:
+        return _AttrGetter, (self.attr_name,)
+
+
+def attribute_keyed_dict(
+    attr_name: str, *, ignore_unpopulated_attribute: bool = False
+) -> Type[KeyFuncDict[Any, Any]]:
+    """A dictionary-based collection type with attribute-based keying.
+
+    .. versionchanged:: 2.0 Renamed :data:`.attribute_mapped_collection` to
+       :func:`.attribute_keyed_dict`.
+
+    Returns a :class:`.KeyFuncDict` factory which will produce new
+    dictionary keys based on the value of a particular named attribute on
+    ORM mapped instances to be added to the dictionary.
+
+    .. note:: the value of the target attribute must be assigned with its
+       value at the time that the object is being added to the
+       dictionary collection.   Additionally, changes to the key attribute
+       are **not tracked**, which means the key in the dictionary is not
+       automatically synchronized with the key value on the target object
+       itself.  See :ref:`key_collections_mutations` for further details.
+
+    .. seealso::
+
+        :ref:`orm_dictionary_collection` - background on use
+
+    :param attr_name: string name of an ORM-mapped attribute
+     on the mapped class, the value of which on a particular instance
+     is to be used as the key for a new dictionary entry for that instance.
+    :param ignore_unpopulated_attribute:  if True, and the target attribute
+     on an object is not populated at all, the operation will be silently
+     skipped.  By default, an error is raised.
+
+     .. versionadded:: 2.0 an error is raised by default if the attribute
+        being used for the dictionary key is determined that it was never
+        populated with any value.  The
+        :paramref:`_orm.attribute_keyed_dict.ignore_unpopulated_attribute`
+        parameter may be set which will instead indicate that this condition
+        should be ignored, and the append operation silently skipped.
+        This is in contrast to the behavior of the 1.x series which would
+        erroneously populate the value in the dictionary with an arbitrary key
+        value of ``None``.
+
+
+    """
+
+    return _mapped_collection_cls(
+        _AttrGetter(attr_name),
+        ignore_unpopulated_attribute=ignore_unpopulated_attribute,
+    )
+
+
+def keyfunc_mapping(
+    keyfunc: Callable[[Any], Any],
+    *,
+    ignore_unpopulated_attribute: bool = False,
+) -> Type[KeyFuncDict[_KT, Any]]:
+    """A dictionary-based collection type with arbitrary keying.
+
+    .. versionchanged:: 2.0 Renamed :data:`.mapped_collection` to
+       :func:`.keyfunc_mapping`.
+
+    Returns a :class:`.KeyFuncDict` factory with a keying function
+    generated from keyfunc, a callable that takes an entity and returns a
+    key value.
+
+    .. note:: the given keyfunc is called only once at the time that the
+       target object is being added to the collection.   Changes to the
+       effective value returned by the function are not tracked.
+
+
+    .. seealso::
+
+        :ref:`orm_dictionary_collection` - background on use
+
+    :param keyfunc: a callable that will be passed the ORM-mapped instance
+     which should then generate a new key to use in the dictionary.
+     If the value returned is :attr:`.LoaderCallableStatus.NO_VALUE`, an error
+     is raised.
+    :param ignore_unpopulated_attribute:  if True, and the callable returns
+     :attr:`.LoaderCallableStatus.NO_VALUE` for a particular instance, the
+     operation will be silently skipped.  By default, an error is raised.
+
+     .. versionadded:: 2.0 an error is raised by default if the callable
+        being used for the dictionary key returns
+        :attr:`.LoaderCallableStatus.NO_VALUE`, which in an ORM attribute
+        context indicates an attribute that was never populated with any value.
+        The :paramref:`_orm.mapped_collection.ignore_unpopulated_attribute`
+        parameter may be set which will instead indicate that this condition
+        should be ignored, and the append operation silently skipped. This is
+        in contrast to the behavior of the 1.x series which would erroneously
+        populate the value in the dictionary with an arbitrary key value of
+        ``None``.
+
+
+    """
+    return _mapped_collection_cls(
+        keyfunc, ignore_unpopulated_attribute=ignore_unpopulated_attribute
+    )
+
+
+class KeyFuncDict(Dict[_KT, _VT]):
+    """Base for ORM mapped dictionary classes.
+
+    Extends the ``dict`` type with additional methods needed by SQLAlchemy ORM
+    collection classes. Use of :class:`_orm.KeyFuncDict` is most directly
+    by using the :func:`.attribute_keyed_dict` or
+    :func:`.column_keyed_dict` class factories.
+    :class:`_orm.KeyFuncDict` may also serve as the base for user-defined
+    custom dictionary classes.
+
+    .. versionchanged:: 2.0 Renamed :class:`.MappedCollection` to
+       :class:`.KeyFuncDict`.
+
+    .. seealso::
+
+        :func:`_orm.attribute_keyed_dict`
+
+        :func:`_orm.column_keyed_dict`
+
+        :ref:`orm_dictionary_collection`
+
+        :ref:`orm_custom_collection`
+
+
+    """
+
+    def __init__(
+        self,
+        keyfunc: Callable[[Any], Any],
+        *dict_args: Any,
+        ignore_unpopulated_attribute: bool = False,
+    ) -> None:
+        """Create a new collection with keying provided by keyfunc.
+
+        keyfunc may be any callable that takes an object and returns an object
+        for use as a dictionary key.
+
+        The keyfunc will be called every time the ORM needs to add a member by
+        value-only (such as when loading instances from the database) or
+        remove a member.  The usual cautions about dictionary keying apply-
+        ``keyfunc(object)`` should return the same output for the life of the
+        collection.  Keying based on mutable properties can result in
+        unreachable instances "lost" in the collection.
+
+        """
+        self.keyfunc = keyfunc
+        self.ignore_unpopulated_attribute = ignore_unpopulated_attribute
+        super().__init__(*dict_args)
+
+    @classmethod
+    def _unreduce(
+        cls,
+        keyfunc: Callable[[Any], Any],
+        values: Dict[_KT, _KT],
+        adapter: Optional[CollectionAdapter] = None,
+    ) -> "KeyFuncDict[_KT, _KT]":
+        mp: KeyFuncDict[_KT, _KT] = KeyFuncDict(keyfunc)
+        mp.update(values)
+        # note that the adapter sets itself up onto this collection
+        # when its `__setstate__` method is called
+        return mp
+
+    def __reduce__(
+        self,
+    ) -> Tuple[
+        Callable[[_KT, _KT], KeyFuncDict[_KT, _KT]],
+        Tuple[Any, Union[Dict[_KT, _KT], Dict[_KT, _KT]], CollectionAdapter],
+    ]:
+        return (
+            KeyFuncDict._unreduce,
+            (
+                self.keyfunc,
+                dict(self),
+                collection_adapter(self),
+            ),
+        )
+
+    @util.preload_module("sqlalchemy.orm.attributes")
+    def _raise_for_unpopulated(
+        self,
+        value: _KT,
+        initiator: Union[AttributeEventToken, Literal[None, False]] = None,
+        *,
+        warn_only: bool,
+    ) -> None:
+        mapper = base.instance_state(value).mapper
+
+        attributes = util.preloaded.orm_attributes
+
+        if not isinstance(initiator, attributes.AttributeEventToken):
+            relationship = "unknown relationship"
+        elif initiator.key in mapper.attrs:
+            relationship = f"{mapper.attrs[initiator.key]}"
+        else:
+            relationship = initiator.key
+
+        if warn_only:
+            util.warn(
+                f"Attribute keyed dictionary value for "
+                f"attribute '{relationship}' was None; this will raise "
+                "in a future release. "
+                f"To skip this assignment entirely, "
+                f'Set the "ignore_unpopulated_attribute=True" '
+                f"parameter on the mapped collection factory."
+            )
+        else:
+            raise sa_exc.InvalidRequestError(
+                "In event triggered from population of "
+                f"attribute '{relationship}' "
+                "(potentially from a backref), "
+                f"can't populate value in KeyFuncDict; "
+                "dictionary key "
+                f"derived from {base.instance_str(value)} is not "
+                f"populated. Ensure appropriate state is set up on "
+                f"the {base.instance_str(value)} object "
+                f"before assigning to the {relationship} attribute. "
+                f"To skip this assignment entirely, "
+                f'Set the "ignore_unpopulated_attribute=True" '
+                f"parameter on the mapped collection factory."
+            )
+
+    @collection.appender  # type: ignore[untyped-decorator]
+    @collection.internally_instrumented  # type: ignore[untyped-decorator]
+    def set(
+        self,
+        value: _KT,
+        _sa_initiator: Union[AttributeEventToken, Literal[None, False]] = None,
+    ) -> None:
+        """Add an item by value, consulting the keyfunc for the key."""
+
+        key = self.keyfunc(value)
+
+        if key is base.NO_VALUE:
+            if not self.ignore_unpopulated_attribute:
+                self._raise_for_unpopulated(
+                    value, _sa_initiator, warn_only=False
+                )
+            else:
+                return
+        elif key is Missing:
+            if not self.ignore_unpopulated_attribute:
+                self._raise_for_unpopulated(
+                    value, _sa_initiator, warn_only=True
+                )
+                key = None
+            else:
+                return
+
+        self.__setitem__(key, value, _sa_initiator)  # type: ignore[call-arg]
+
+    @collection.remover  # type: ignore[untyped-decorator]
+    @collection.internally_instrumented  # type: ignore[untyped-decorator]
+    def remove(
+        self,
+        value: _KT,
+        _sa_initiator: Union[AttributeEventToken, Literal[None, False]] = None,
+    ) -> None:
+        """Remove an item by value, consulting the keyfunc for the key."""
+
+        key = self.keyfunc(value)
+
+        if key is base.NO_VALUE:
+            if not self.ignore_unpopulated_attribute:
+                self._raise_for_unpopulated(
+                    value, _sa_initiator, warn_only=False
+                )
+            return
+        elif key is Missing:
+            if not self.ignore_unpopulated_attribute:
+                self._raise_for_unpopulated(
+                    value, _sa_initiator, warn_only=True
+                )
+                key = None
+            else:
+                return
+
+        # Let self[key] raise if key is not in this collection
+        # testlib.pragma exempt:__ne__
+        if self[key] != value:
+            raise sa_exc.InvalidRequestError(
+                "Can not remove '%s': collection holds '%s' for key '%s'. "
+                "Possible cause: is the KeyFuncDict key function "
+                "based on mutable properties or properties that only obtain "
+                "values after flush?" % (value, self[key], key)
+            )
+        self.__delitem__(key, _sa_initiator)  # type: ignore[call-arg]
+
+
+def _mapped_collection_cls(
+    keyfunc: Callable[[Any], Any], ignore_unpopulated_attribute: bool
+) -> Type[KeyFuncDict[_KT, _KT]]:
+    class _MKeyfuncMapped(KeyFuncDict[_KT, _KT]):
+        def __init__(self, *dict_args: Any) -> None:
+            super().__init__(
+                keyfunc,
+                *dict_args,
+                ignore_unpopulated_attribute=ignore_unpopulated_attribute,
+            )
+
+    return _MKeyfuncMapped
+
+
+MappedCollection = KeyFuncDict
+"""A synonym for :class:`.KeyFuncDict`.
+
+.. versionchanged:: 2.0 Renamed :class:`.MappedCollection` to
+   :class:`.KeyFuncDict`.
+
+"""
+
+mapped_collection = keyfunc_mapping
+"""A synonym for :func:`_orm.keyfunc_mapping`.
+
+.. versionchanged:: 2.0 Renamed :data:`.mapped_collection` to
+   :func:`_orm.keyfunc_mapping`
+
+"""
+
+attribute_mapped_collection = attribute_keyed_dict
+"""A synonym for :func:`_orm.attribute_keyed_dict`.
+
+.. versionchanged:: 2.0 Renamed :data:`.attribute_mapped_collection` to
+   :func:`_orm.attribute_keyed_dict`
+
+"""
+
+column_mapped_collection = column_keyed_dict
+"""A synonym for :func:`_orm.column_keyed_dict.
+
+.. versionchanged:: 2.0 Renamed :func:`.column_mapped_collection` to
+   :func:`_orm.column_keyed_dict`
+
+"""

venv/Lib/site-packages/sqlalchemy/orm/path_registry.py

@@ -0,0 +1,824 @@
+# orm/path_registry.py
+# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
+"""Path tracking utilities, representing mapper graph traversals."""
+
+from __future__ import annotations
+
+from functools import reduce
+from itertools import chain
+import logging
+import operator
+from typing import Any
+from typing import cast
+from typing import Dict
+from typing import Iterator
+from typing import List
+from typing import Optional
+from typing import overload
+from typing import Sequence
+from typing import Tuple
+from typing import TYPE_CHECKING
+from typing import Union
+
+from . import base as orm_base
+from ._typing import insp_is_mapper_property
+from .. import exc
+from .. import util
+from ..sql import visitors
+from ..sql.cache_key import HasCacheKey
+
+if TYPE_CHECKING:
+    from ._typing import _InternalEntityType
+    from .interfaces import StrategizedProperty
+    from .mapper import Mapper
+    from .relationships import RelationshipProperty
+    from .util import AliasedInsp
+    from ..sql.cache_key import _CacheKeyTraversalType
+    from ..sql.elements import BindParameter
+    from ..sql.visitors import anon_map
+    from ..util.typing import _LiteralStar
+    from ..util.typing import TypeGuard
+
+    def is_root(path: PathRegistry) -> TypeGuard[RootRegistry]: ...
+
+    def is_entity(path: PathRegistry) -> TypeGuard[AbstractEntityRegistry]: ...
+
+else:
+    is_root = operator.attrgetter("is_root")
+    is_entity = operator.attrgetter("is_entity")
+
+
+_SerializedPath = List[Any]
+_StrPathToken = str
+_PathElementType = Union[
+    _StrPathToken, "_InternalEntityType[Any]", "StrategizedProperty[Any]"
+]
+
+# the representation is in fact
+# a tuple with alternating:
+# [_InternalEntityType[Any], Union[str, StrategizedProperty[Any]],
+# _InternalEntityType[Any], Union[str, StrategizedProperty[Any]], ...]
+# this might someday be a tuple of 2-tuples instead, but paths can be
+# chopped at odd intervals as well so this is less flexible
+_PathRepresentation = Tuple[_PathElementType, ...]
+
+# NOTE: these names are weird since the array is 0-indexed,
+# the "_Odd" entries are at 0, 2, 4, etc
+_OddPathRepresentation = Sequence["_InternalEntityType[Any]"]
+_EvenPathRepresentation = Sequence[Union["StrategizedProperty[Any]", str]]
+
+
+log = logging.getLogger(__name__)
+
+
+def _unreduce_path(path: _SerializedPath) -> PathRegistry:
+    return PathRegistry.deserialize(path)
+
+
+_WILDCARD_TOKEN: _LiteralStar = "*"
+_DEFAULT_TOKEN = "_sa_default"
+
+
+class PathRegistry(HasCacheKey):
+    """Represent query load paths and registry functions.
+
+    Basically represents structures like:
+
+    (<User mapper>, "orders", <Order mapper>, "items", <Item mapper>)
+
+    These structures are generated by things like
+    query options (joinedload(), subqueryload(), etc.) and are
+    used to compose keys stored in the query._attributes dictionary
+    for various options.
+
+    They are then re-composed at query compile/result row time as
+    the query is formed and as rows are fetched, where they again
+    serve to compose keys to look up options in the context.attributes
+    dictionary, which is copied from query._attributes.
+
+    The path structure has a limited amount of caching, where each
+    "root" ultimately pulls from a fixed registry associated with
+    the first mapper, that also contains elements for each of its
+    property keys.  However paths longer than two elements, which
+    are the exception rather than the rule, are generated on an
+    as-needed basis.
+
+    """
+
+    __slots__ = ()
+
+    is_token = False
+    is_root = False
+    has_entity = False
+    is_property = False
+    is_entity = False
+
+    is_unnatural: bool
+
+    path: _PathRepresentation
+    natural_path: _PathRepresentation
+    parent: Optional[PathRegistry]
+    root: RootRegistry
+
+    _cache_key_traversal: _CacheKeyTraversalType = [
+        ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key_list)
+    ]
+
+    def __eq__(self, other: Any) -> bool:
+        try:
+            return other is not None and self.path == other._path_for_compare
+        except AttributeError:
+            util.warn(
+                "Comparison of PathRegistry to %r is not supported"
+                % (type(other))
+            )
+            return False
+
+    def __ne__(self, other: Any) -> bool:
+        try:
+            return other is None or self.path != other._path_for_compare
+        except AttributeError:
+            util.warn(
+                "Comparison of PathRegistry to %r is not supported"
+                % (type(other))
+            )
+            return True
+
+    @property
+    def _path_for_compare(self) -> Optional[_PathRepresentation]:
+        return self.path
+
+    def odd_element(self, index: int) -> _InternalEntityType[Any]:
+        return self.path[index]  # type: ignore
+
+    def set(self, attributes: Dict[Any, Any], key: Any, value: Any) -> None:
+        log.debug("set '%s' on path '%s' to '%s'", key, self, value)
+        attributes[(key, self.natural_path)] = value
+
+    def setdefault(
+        self, attributes: Dict[Any, Any], key: Any, value: Any
+    ) -> None:
+        log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value)
+        attributes.setdefault((key, self.natural_path), value)
+
+    def get(
+        self, attributes: Dict[Any, Any], key: Any, value: Optional[Any] = None
+    ) -> Any:
+        key = (key, self.natural_path)
+        if key in attributes:
+            return attributes[key]
+        else:
+            return value
+
+    def __len__(self) -> int:
+        return len(self.path)
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    @overload
+    def __getitem__(self, entity: _StrPathToken) -> TokenRegistry: ...
+
+    @overload
+    def __getitem__(self, entity: int) -> _PathElementType: ...
+
+    @overload
+    def __getitem__(self, entity: slice) -> _PathRepresentation: ...
+
+    @overload
+    def __getitem__(
+        self, entity: _InternalEntityType[Any]
+    ) -> AbstractEntityRegistry: ...
+
+    @overload
+    def __getitem__(
+        self, entity: StrategizedProperty[Any]
+    ) -> PropRegistry: ...
+
+    def __getitem__(
+        self,
+        entity: Union[
+            _StrPathToken,
+            int,
+            slice,
+            _InternalEntityType[Any],
+            StrategizedProperty[Any],
+        ],
+    ) -> Union[
+        TokenRegistry,
+        _PathElementType,
+        _PathRepresentation,
+        PropRegistry,
+        AbstractEntityRegistry,
+    ]:
+        raise NotImplementedError()
+
+    # TODO: what are we using this for?
+    @property
+    def length(self) -> int:
+        return len(self.path)
+
+    def pairs(
+        self,
+    ) -> Iterator[
+        Tuple[_InternalEntityType[Any], Union[str, StrategizedProperty[Any]]]
+    ]:
+        odd_path = cast(_OddPathRepresentation, self.path)
+        even_path = cast(_EvenPathRepresentation, odd_path)
+        for i in range(0, len(odd_path), 2):
+            yield odd_path[i], even_path[i + 1]
+
+    def contains_mapper(self, mapper: Mapper[Any]) -> bool:
+        _m_path = cast(_OddPathRepresentation, self.path)
+        for path_mapper in [_m_path[i] for i in range(0, len(_m_path), 2)]:
+            if path_mapper.mapper.isa(mapper):
+                return True
+        else:
+            return False
+
+    def contains(self, attributes: Dict[Any, Any], key: Any) -> bool:
+        return (key, self.path) in attributes
+
+    def __reduce__(self) -> Any:
+        return _unreduce_path, (self.serialize(),)
+
+    @classmethod
+    def _serialize_path(cls, path: _PathRepresentation) -> _SerializedPath:
+        _m_path = cast(_OddPathRepresentation, path)
+        _p_path = cast(_EvenPathRepresentation, path)
+
+        return list(
+            zip(
+                tuple(
+                    m.class_ if (m.is_mapper or m.is_aliased_class) else str(m)
+                    for m in [_m_path[i] for i in range(0, len(_m_path), 2)]
+                ),
+                tuple(
+                    p.key if insp_is_mapper_property(p) else str(p)
+                    for p in [_p_path[i] for i in range(1, len(_p_path), 2)]
+                )
+                + (None,),
+            )
+        )
+
+    @classmethod
+    def _deserialize_path(cls, path: _SerializedPath) -> _PathRepresentation:
+        def _deserialize_mapper_token(mcls: Any) -> Any:
+            return (
+                # note: we likely dont want configure=True here however
+                # this is maintained at the moment for backwards compatibility
+                orm_base._inspect_mapped_class(mcls, configure=True)
+                if mcls not in PathToken._intern
+                else PathToken._intern[mcls]
+            )
+
+        def _deserialize_key_token(mcls: Any, key: Any) -> Any:
+            if key is None:
+                return None
+            elif key in PathToken._intern:
+                return PathToken._intern[key]
+            else:
+                mp = orm_base._inspect_mapped_class(mcls, configure=True)
+                assert mp is not None
+                return mp.attrs[key]
+
+        p = tuple(
+            chain(
+                *[
+                    (
+                        _deserialize_mapper_token(mcls),
+                        _deserialize_key_token(mcls, key),
+                    )
+                    for mcls, key in path
+                ]
+            )
+        )
+        if p and p[-1] is None:
+            p = p[0:-1]
+        return p
+
+    def serialize(self) -> _SerializedPath:
+        path = self.path
+        return self._serialize_path(path)
+
+    @classmethod
+    def deserialize(cls, path: _SerializedPath) -> PathRegistry:
+        assert path is not None
+        p = cls._deserialize_path(path)
+        return cls.coerce(p)
+
+    @overload
+    @classmethod
+    def per_mapper(cls, mapper: Mapper[Any]) -> CachingEntityRegistry: ...
+
+    @overload
+    @classmethod
+    def per_mapper(cls, mapper: AliasedInsp[Any]) -> SlotsEntityRegistry: ...
+
+    @classmethod
+    def per_mapper(
+        cls, mapper: _InternalEntityType[Any]
+    ) -> AbstractEntityRegistry:
+        if mapper.is_mapper:
+            return CachingEntityRegistry(cls.root, mapper)
+        else:
+            return SlotsEntityRegistry(cls.root, mapper)
+
+    @classmethod
+    def coerce(cls, raw: _PathRepresentation) -> PathRegistry:
+        def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
+            return prev[next_]
+
+        # can't quite get mypy to appreciate this one :)
+        return reduce(_red, raw, cls.root)  # type: ignore
+
+    def __add__(self, other: PathRegistry) -> PathRegistry:
+        def _red(prev: PathRegistry, next_: _PathElementType) -> PathRegistry:
+            return prev[next_]
+
+        return reduce(_red, other.path, self)
+
+    def __str__(self) -> str:
+        return f"ORM Path[{' -> '.join(str(elem) for elem in self.path)}]"
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}({self.path!r})"
+
+
+class CreatesToken(PathRegistry):
+    __slots__ = ()
+
+    is_aliased_class: bool
+    is_root: bool
+
+    def token(self, token: _StrPathToken) -> TokenRegistry:
+        if token.endswith(f":{_WILDCARD_TOKEN}"):
+            return TokenRegistry(self, token)
+        elif token.endswith(f":{_DEFAULT_TOKEN}"):
+            return TokenRegistry(self.root, token)
+        else:
+            raise exc.ArgumentError(f"invalid token: {token}")
+
+
+class RootRegistry(CreatesToken):
+    """Root registry, defers to mappers so that
+    paths are maintained per-root-mapper.
+
+    """
+
+    __slots__ = ()
+
+    inherit_cache = True
+
+    path = natural_path = ()
+    has_entity = False
+    is_aliased_class = False
+    is_root = True
+    is_unnatural = False
+
+    def _getitem(
+        self, entity: Any
+    ) -> Union[TokenRegistry, AbstractEntityRegistry]:
+        if entity in PathToken._intern:
+            if TYPE_CHECKING:
+                assert isinstance(entity, _StrPathToken)
+            return TokenRegistry(self, PathToken._intern[entity])
+        else:
+            try:
+                return entity._path_registry  # type: ignore
+            except AttributeError:
+                raise IndexError(
+                    f"invalid argument for RootRegistry.__getitem__: {entity}"
+                )
+
+    def _truncate_recursive(self) -> RootRegistry:
+        return self
+
+    if not TYPE_CHECKING:
+        __getitem__ = _getitem
+
+
+PathRegistry.root = RootRegistry()
+
+
+class PathToken(orm_base.InspectionAttr, HasCacheKey, str):
+    """cacheable string token"""
+
+    _intern: Dict[str, PathToken] = {}
+
+    def _gen_cache_key(
+        self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
+    ) -> Tuple[Any, ...]:
+        return (str(self),)
+
+    @property
+    def _path_for_compare(self) -> Optional[_PathRepresentation]:
+        return None
+
+    @classmethod
+    def intern(cls, strvalue: str) -> PathToken:
+        if strvalue in cls._intern:
+            return cls._intern[strvalue]
+        else:
+            cls._intern[strvalue] = result = PathToken(strvalue)
+            return result
+
+
+class TokenRegistry(PathRegistry):
+    __slots__ = ("token", "parent", "path", "natural_path")
+
+    inherit_cache = True
+
+    token: _StrPathToken
+    parent: CreatesToken
+
+    def __init__(self, parent: CreatesToken, token: _StrPathToken):
+        token = PathToken.intern(token)
+
+        self.token = token
+        self.parent = parent
+        self.path = parent.path + (token,)
+        self.natural_path = parent.natural_path + (token,)
+
+    has_entity = False
+
+    is_token = True
+
+    def generate_for_superclasses(self) -> Iterator[PathRegistry]:
+        # NOTE: this method is no longer used.  consider removal
+        parent = self.parent
+        if is_root(parent):
+            yield self
+            return
+
+        if TYPE_CHECKING:
+            assert isinstance(parent, AbstractEntityRegistry)
+        if not parent.is_aliased_class:
+            for mp_ent in parent.mapper.iterate_to_root():
+                yield TokenRegistry(parent.parent[mp_ent], self.token)
+        elif (
+            parent.is_aliased_class
+            and cast(
+                "AliasedInsp[Any]",
+                parent.entity,
+            )._is_with_polymorphic
+        ):
+            yield self
+            for ent in cast(
+                "AliasedInsp[Any]", parent.entity
+            )._with_polymorphic_entities:
+                yield TokenRegistry(parent.parent[ent], self.token)
+        else:
+            yield self
+
+    def _generate_natural_for_superclasses(
+        self,
+    ) -> Iterator[_PathRepresentation]:
+        parent = self.parent
+        if is_root(parent):
+            yield self.natural_path
+            return
+
+        if TYPE_CHECKING:
+            assert isinstance(parent, AbstractEntityRegistry)
+        for mp_ent in parent.mapper.iterate_to_root():
+            yield TokenRegistry(parent.parent[mp_ent], self.token).natural_path
+        if (
+            parent.is_aliased_class
+            and cast(
+                "AliasedInsp[Any]",
+                parent.entity,
+            )._is_with_polymorphic
+        ):
+            yield self.natural_path
+            for ent in cast(
+                "AliasedInsp[Any]", parent.entity
+            )._with_polymorphic_entities:
+                yield (
+                    TokenRegistry(parent.parent[ent], self.token).natural_path
+                )
+        else:
+            yield self.natural_path
+
+    def _getitem(self, entity: Any) -> Any:
+        try:
+            return self.path[entity]
+        except TypeError as err:
+            raise IndexError(f"{entity}") from err
+
+    if not TYPE_CHECKING:
+        __getitem__ = _getitem
+
+
+class PropRegistry(PathRegistry):
+    __slots__ = (
+        "prop",
+        "parent",
+        "path",
+        "natural_path",
+        "has_entity",
+        "entity",
+        "mapper",
+        "_wildcard_path_loader_key",
+        "_default_path_loader_key",
+        "_loader_key",
+        "is_unnatural",
+    )
+    inherit_cache = True
+    is_property = True
+
+    prop: StrategizedProperty[Any]
+    mapper: Optional[Mapper[Any]]
+    entity: Optional[_InternalEntityType[Any]]
+
+    def __init__(
+        self, parent: AbstractEntityRegistry, prop: StrategizedProperty[Any]
+    ):
+
+        # restate this path in terms of the
+        # given StrategizedProperty's parent.
+        insp = cast("_InternalEntityType[Any]", parent[-1])
+        natural_parent: AbstractEntityRegistry = parent
+
+        # inherit "is_unnatural" from the parent
+        self.is_unnatural = parent.parent.is_unnatural or bool(
+            parent.mapper.inherits
+        )
+
+        if not insp.is_aliased_class or insp._use_mapper_path:  # type: ignore
+            parent = natural_parent = parent.parent[prop.parent]
+        elif (
+            insp.is_aliased_class
+            and insp.with_polymorphic_mappers
+            and prop.parent in insp.with_polymorphic_mappers
+        ):
+            subclass_entity: _InternalEntityType[Any] = parent[-1]._entity_for_mapper(prop.parent)  # type: ignore  # noqa: E501
+            parent = parent.parent[subclass_entity]
+
+            # when building a path where with_polymorphic() is in use,
+            # special logic to determine the "natural path" when subclass
+            # entities are used.
+            #
+            # here we are trying to distinguish between a path that starts
+            # on a with_polymorphic entity vs. one that starts on a
+            # normal entity that introduces a with_polymorphic() in the
+            # middle using of_type():
+            #
+            #  # as in test_polymorphic_rel->
+            #  #    test_subqueryload_on_subclass_uses_path_correctly
+            #  wp = with_polymorphic(RegularEntity, "*")
+            #  sess.query(wp).options(someload(wp.SomeSubEntity.foos))
+            #
+            # vs
+            #
+            #  # as in test_relationship->JoinedloadWPolyOfTypeContinued
+            #  wp = with_polymorphic(SomeFoo, "*")
+            #  sess.query(RegularEntity).options(
+            #       someload(RegularEntity.foos.of_type(wp))
+            #       .someload(wp.SubFoo.bar)
+            #   )
+            #
+            # in the former case, the Query as it generates a path that we
+            # want to match will be in terms of the with_polymorphic at the
+            # beginning.  in the latter case, Query will generate simple
+            # paths that don't know about this with_polymorphic, so we must
+            # use a separate natural path.
+            #
+            #
+            if parent.parent:
+                natural_parent = parent.parent[subclass_entity.mapper]
+                self.is_unnatural = True
+            else:
+                natural_parent = parent
+        elif (
+            natural_parent.parent
+            and insp.is_aliased_class
+            and prop.parent  # this should always be the case here
+            is not insp.mapper
+            and insp.mapper.isa(prop.parent)
+        ):
+            natural_parent = parent.parent[prop.parent]
+
+        self.prop = prop
+        self.parent = parent
+        self.path = parent.path + (prop,)
+        self.natural_path = natural_parent.natural_path + (prop,)
+
+        self.has_entity = prop._links_to_entity
+        if prop._is_relationship:
+            if TYPE_CHECKING:
+                assert isinstance(prop, RelationshipProperty)
+            self.entity = prop.entity
+            self.mapper = prop.mapper
+        else:
+            self.entity = None
+            self.mapper = None
+
+        self._wildcard_path_loader_key = (
+            "loader",
+            parent.natural_path + self.prop._wildcard_token,
+        )
+        self._default_path_loader_key = self.prop._default_path_loader_key
+        self._loader_key = ("loader", self.natural_path)
+
+    def _truncate_recursive(self) -> PropRegistry:
+        earliest = None
+        for i, token in enumerate(reversed(self.path[:-1])):
+            if token is self.prop:
+                earliest = i
+
+        if earliest is None:
+            return self
+        else:
+            return self.coerce(self.path[0 : -(earliest + 1)])  # type: ignore
+
+    @property
+    def entity_path(self) -> AbstractEntityRegistry:
+        assert self.entity is not None
+        return self[self.entity]
+
+    def _getitem(
+        self, entity: Union[int, slice, _InternalEntityType[Any]]
+    ) -> Union[AbstractEntityRegistry, _PathElementType, _PathRepresentation]:
+        if isinstance(entity, (int, slice)):
+            return self.path[entity]
+        else:
+            return SlotsEntityRegistry(self, entity)
+
+    if not TYPE_CHECKING:
+        __getitem__ = _getitem
+
+
+class AbstractEntityRegistry(CreatesToken):
+    __slots__ = (
+        "key",
+        "parent",
+        "is_aliased_class",
+        "path",
+        "entity",
+        "natural_path",
+    )
+
+    has_entity = True
+    is_entity = True
+
+    parent: Union[RootRegistry, PropRegistry]
+    key: _InternalEntityType[Any]
+    entity: _InternalEntityType[Any]
+    is_aliased_class: bool
+
+    def __init__(
+        self,
+        parent: Union[RootRegistry, PropRegistry],
+        entity: _InternalEntityType[Any],
+    ):
+        self.key = entity
+        self.parent = parent
+        self.is_aliased_class = entity.is_aliased_class
+        self.entity = entity
+        self.path = parent.path + (entity,)
+
+        # the "natural path" is the path that we get when Query is traversing
+        # from the lead entities into the various relationships; it corresponds
+        # to the structure of mappers and relationships. when we are given a
+        # path that comes from loader options, as of 1.3 it can have ac-hoc
+        # with_polymorphic() and other AliasedInsp objects inside of it, which
+        # are usually not present in mappings.  So here we track both the
+        # "enhanced" path in self.path and the "natural" path that doesn't
+        # include those objects so these two traversals can be matched up.
+
+        # the test here for "(self.is_aliased_class or parent.is_unnatural)"
+        # are to avoid the more expensive conditional logic that follows if we
+        # know we don't have to do it.   This conditional can just as well be
+        # "if parent.path:", it just is more function calls.
+        #
+        # This is basically the only place that the "is_unnatural" flag
+        # actually changes behavior.
+        if parent.path and (self.is_aliased_class or parent.is_unnatural):
+            # this is an infrequent code path used for loader strategies that
+            # also make use of of_type() or other intricate polymorphic
+            # base/subclass combinations
+            parent_natural_entity = parent.natural_path[-1]
+
+            if entity.mapper.isa(
+                parent_natural_entity.mapper  # type: ignore
+            ) or parent_natural_entity.mapper.isa(  # type: ignore
+                entity.mapper
+            ):
+                # when the entity mapper and parent mapper are in an
+                # inheritance relationship, use entity.mapper in natural_path.
+                # First case: entity.mapper inherits from parent mapper (e.g.,
+                # accessing a subclass mapper through parent path). Second case
+                # (issue #13193): parent mapper inherits from entity.mapper
+                # (e.g., parent path has Sub(Base) but we're accessing with
+                # Base where Base.related is declared, so use Base in
+                # natural_path).
+                self.natural_path = parent.natural_path + (entity.mapper,)
+            else:
+                self.natural_path = parent.natural_path + (
+                    parent_natural_entity.entity,  # type: ignore
+                )
+        # it seems to make sense that since these paths get mixed up
+        # with statements that are cached or not, we should make
+        # sure the natural path is cacheable across different occurrences
+        # of equivalent AliasedClass objects.  however, so far this
+        # does not seem to be needed for whatever reason.
+        # elif not parent.path and self.is_aliased_class:
+        #     self.natural_path = (self.entity._generate_cache_key()[0], )
+        else:
+            self.natural_path = self.path
+
+    def _truncate_recursive(self) -> AbstractEntityRegistry:
+        return self.parent._truncate_recursive()[self.entity]
+
+    @property
+    def root_entity(self) -> _InternalEntityType[Any]:
+        return self.odd_element(0)
+
+    @property
+    def entity_path(self) -> PathRegistry:
+        return self
+
+    @property
+    def mapper(self) -> Mapper[Any]:
+        return self.entity.mapper
+
+    def __bool__(self) -> bool:
+        return True
+
+    def _getitem(
+        self, entity: Any
+    ) -> Union[_PathElementType, _PathRepresentation, PathRegistry]:
+        if isinstance(entity, (int, slice)):
+            return self.path[entity]
+        elif entity in PathToken._intern:
+            return TokenRegistry(self, PathToken._intern[entity])
+        else:
+            return PropRegistry(self, entity)
+
+    if not TYPE_CHECKING:
+        __getitem__ = _getitem
+
+
+class SlotsEntityRegistry(AbstractEntityRegistry):
+    # for aliased class, return lightweight, no-cycles created
+    # version
+    inherit_cache = True
+
+
+class _ERDict(Dict[Any, Any]):
+    def __init__(self, registry: CachingEntityRegistry):
+        self.registry = registry
+
+    def __missing__(self, key: Any) -> PropRegistry:
+        self[key] = item = PropRegistry(self.registry, key)
+
+        return item
+
+
+class CachingEntityRegistry(AbstractEntityRegistry):
+    # for long lived mapper, return dict based caching
+    # version that creates reference cycles
+
+    __slots__ = ("_cache",)
+
+    inherit_cache = True
+
+    def __init__(
+        self,
+        parent: Union[RootRegistry, PropRegistry],
+        entity: _InternalEntityType[Any],
+    ):
+        super().__init__(parent, entity)
+        self._cache = _ERDict(self)
+
+    def pop(self, key: Any, default: Any) -> Any:
+        return self._cache.pop(key, default)
+
+    def _getitem(self, entity: Any) -> Any:
+        if isinstance(entity, (int, slice)):
+            return self.path[entity]
+        elif isinstance(entity, PathToken):
+            return TokenRegistry(self, entity)
+        else:
+            return self._cache[entity]
+
+    if not TYPE_CHECKING:
+        __getitem__ = _getitem
+
+
+if TYPE_CHECKING:
+
+    def path_is_entity(
+        path: PathRegistry,
+    ) -> TypeGuard[AbstractEntityRegistry]: ...
+
+    def path_is_property(path: PathRegistry) -> TypeGuard[PropRegistry]: ...
+
+else:
+    path_is_entity = operator.attrgetter("is_entity")
+    path_is_property = operator.attrgetter("is_property")