Kobe/VerpotJeLot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lol

Browse Source
This commit is contained in:
Kobe
2025-05-22 20:25:38 +02:00
parent 09f6750c2b
commit ce03fbf12f
529 changed files with 3353 additions and 3312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
venv/Lib/site-packages/sqlalchemy/orm/strategy_options.py
View File

@@ -34,6 +34,7 @@ from .attributes import QueryableAttribute
from .base import InspectionAttr
from .interfaces import LoaderOption
from .path_registry import _DEFAULT_TOKEN
from .path_registry import _StrPathToken
from .path_registry import _WILDCARD_TOKEN
from .path_registry import AbstractEntityRegistry
from .path_registry import path_is_property
@@ -77,7 +78,7 @@ if typing.TYPE_CHECKING:
from ..sql.cache_key import CacheKey
_AttrType = Union[str, "QueryableAttribute[Any]"]
_AttrType = Union[Literal["*"], "QueryableAttribute[Any]"]
_WildcardKeyType = Literal["relationship", "column"]
_StrategySpec = Dict[str, Any]
@@ -541,7 +542,12 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
)
def defaultload(self, attr: _AttrType) -> Self:
"""Indicate an attribute should load using its default loader style.
"""Indicate an attribute should load using its predefined loader style.
The behavior of this loading option is to not change the current
loading style of the attribute, meaning that the previously configured
one is used or, if no previous style was selected, the default
loading will be used.
This method is used to link to other loader options further into
a chain of attributes without altering the loader style of the links
@@ -741,7 +747,7 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
)
return self._set_column_strategy(
(key,), {"query_expression": True}, opts={"expression": expression}
(key,), {"query_expression": True}, extra_criteria=(expression,)
)
def selectin_polymorphic(self, classes: Iterable[Type[Any]]) -> Self:
@@ -813,6 +819,7 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
attrs: Tuple[_AttrType, ...],
strategy: Optional[_StrategySpec],
opts: Optional[_OptsType] = None,
extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
strategy_key = self._coerce_strat(strategy)
@@ -822,6 +829,7 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
_COLUMN_TOKEN,
opts=opts,
attr_group=attrs,
extra_criteria=extra_criteria,
)
return self
@@ -878,6 +886,7 @@ class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
raise NotImplementedError()
@@ -977,6 +986,7 @@ class Load(_AbstractLoad):
__slots__ = (
"path",
"context",
"additional_source_entities",
)
_traverse_internals = [
@@ -986,11 +996,16 @@ class Load(_AbstractLoad):
visitors.InternalTraversal.dp_has_cache_key_list,
),
("propagate_to_loaders", visitors.InternalTraversal.dp_boolean),
(
"additional_source_entities",
visitors.InternalTraversal.dp_has_cache_key_list,
),
]
_cache_key_traversal = None
path: PathRegistry
context: Tuple[_LoadElement, ...]
additional_source_entities: Tuple[_InternalEntityType[Any], ...]
def __init__(self, entity: _EntityType[Any]):
insp = cast("Union[Mapper[Any], AliasedInsp[Any]]", inspect(entity))
@@ -999,16 +1014,20 @@ class Load(_AbstractLoad):
self.path = insp._path_registry
self.context = ()
self.propagate_to_loaders = False
self.additional_source_entities = ()
def __str__(self) -> str:
return f"Load({self.path[0]})"
@classmethod
def _construct_for_existing_path(cls, path: PathRegistry) -> Load:
def _construct_for_existing_path(
cls, path: AbstractEntityRegistry
) -> Load:
load = cls.__new__(cls)
load.path = path
load.context = ()
load.propagate_to_loaders = False
load.additional_source_entities = ()
return load
def _adapt_cached_option_to_uncached_option(
@@ -1016,6 +1035,13 @@ class Load(_AbstractLoad):
) -> ORMOption:
return self._adjust_for_extra_criteria(context)
def _prepend_path(self, path: PathRegistry) -> Load:
cloned = self._clone()
cloned.context = tuple(
element._prepend_path(path) for element in self.context
)
return cloned
def _adjust_for_extra_criteria(self, context: QueryContext) -> Load:
"""Apply the current bound parameters in a QueryContext to all
occurrences "extra_criteria" stored within this ``Load`` object,
@@ -1029,16 +1055,10 @@ class Load(_AbstractLoad):
found_crit = False
def process(opt: _LoadElement) -> _LoadElement:
if not opt._extra_criteria:
return opt
nonlocal orig_cache_key, replacement_cache_key, found_crit
found_crit = True
# avoid generating cache keys for the queries if we don't
# actually have any extra_criteria options, which is the
# common case
if orig_cache_key is None or replacement_cache_key is None:
orig_cache_key = orig_query._generate_cache_key()
replacement_cache_key = context.query._generate_cache_key()
@@ -1052,8 +1072,12 @@ class Load(_AbstractLoad):
)
for crit in opt._extra_criteria
)
return opt
# avoid generating cache keys for the queries if we don't
# actually have any extra_criteria options, which is the
# common case
new_context = tuple(
process(value._clone()) if value._extra_criteria else value
for value in self.context
@@ -1116,9 +1140,12 @@ class Load(_AbstractLoad):
assert cloned.propagate_to_loaders == self.propagate_to_loaders
if not orm_util._entity_corresponds_to_use_path_impl(
cast("_InternalEntityType[Any]", parent.path[-1]),
cast("_InternalEntityType[Any]", cloned.path[0]),
if not any(
orm_util._entity_corresponds_to_use_path_impl(
elem, cloned.path.odd_element(0)
)
for elem in (parent.path.odd_element(-1),)
+ parent.additional_source_entities
):
if len(cloned.path) > 1:
attrname = cloned.path[1]
@@ -1137,6 +1164,9 @@ class Load(_AbstractLoad):
if cloned.context:
parent.context += cloned.context
parent.additional_source_entities += (
cloned.additional_source_entities
)
@_generative
def options(self, *opts: _AbstractLoad) -> Self:
@@ -1191,6 +1221,7 @@ class Load(_AbstractLoad):
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
# for individual strategy that needs to propagate, set the whole
# Load container to also propagate, so that it shows up in
@@ -1224,9 +1255,14 @@ class Load(_AbstractLoad):
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
extra_criteria=extra_criteria,
)
if load_element:
self.context += (load_element,)
assert opts is not None
self.additional_source_entities += cast(
"Tuple[_InternalEntityType[Any]]", opts["entities"]
)
else:
for attr in attrs:
@@ -1240,6 +1276,7 @@ class Load(_AbstractLoad):
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
extra_criteria=extra_criteria,
)
else:
load_element = _AttributeStrategyLoad.create(
@@ -1251,6 +1288,7 @@ class Load(_AbstractLoad):
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
extra_criteria=extra_criteria,
)
if load_element:
@@ -1314,6 +1352,7 @@ class _WildcardLoad(_AbstractLoad):
attr_group=None,
propagate_to_loaders=True,
reconcile_to_other=None,
extra_criteria=None,
):
assert attrs is not None
attr = attrs[0]
@@ -1330,6 +1369,8 @@ class _WildcardLoad(_AbstractLoad):
if opts:
self.local_opts = util.immutabledict(opts)
assert extra_criteria is None
def options(self, *opts: _AbstractLoad) -> Self:
raise NotImplementedError("Star option does not support sub-options")
@@ -1616,7 +1657,9 @@ class _LoadElement(
return effective_path
def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
def _init_path(
self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
):
"""Apply ORM attributes and/or wildcard to an existing path, producing
a new path.
@@ -1668,7 +1711,7 @@ class _LoadElement(
def create(
cls,
path: PathRegistry,
attr: Optional[_AttrType],
attr: Union[_AttrType, _StrPathToken, None],
strategy: Optional[_StrategyKey],
wildcard_key: Optional[_WildcardKeyType],
local_opts: Optional[_OptsType],
@@ -1676,6 +1719,7 @@ class _LoadElement(
raiseerr: bool = True,
attr_group: Optional[_AttrGroupType] = None,
reconcile_to_other: Optional[bool] = None,
extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> _LoadElement:
"""Create a new :class:`._LoadElement` object."""
@@ -1695,7 +1739,9 @@ class _LoadElement(
else:
opt._reconcile_to_other = None
path = opt._init_path(path, attr, wildcard_key, attr_group, raiseerr)
path = opt._init_path(
path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
)
if not path:
return None # type: ignore
@@ -1714,9 +1760,7 @@ class _LoadElement(
return cloned
def _prepend_path_from(
self, parent: Union[Load, _LoadElement]
) -> _LoadElement:
def _prepend_path_from(self, parent: Load) -> _LoadElement:
"""adjust the path of this :class:`._LoadElement` to be
a subpath of that of the given parent :class:`_orm.Load` object's
path.
@@ -1725,22 +1769,30 @@ class _LoadElement(
which is in turn part of the :meth:`_orm.Load.options` method.
"""
if not any(
orm_util._entity_corresponds_to_use_path_impl(
elem,
self.path.odd_element(0),
)
for elem in (parent.path.odd_element(-1),)
+ parent.additional_source_entities
):
raise sa_exc.ArgumentError(
f'Attribute "{self.path[1]}" does not link '
f'from element "{parent.path[-1]}".'
)
return self._prepend_path(parent.path)
def _prepend_path(self, path: PathRegistry) -> _LoadElement:
cloned = self._clone()
assert cloned.strategy == self.strategy
assert cloned.local_opts == self.local_opts
assert cloned.is_class_strategy == self.is_class_strategy
if not orm_util._entity_corresponds_to_use_path_impl(
cast("_InternalEntityType[Any]", parent.path[-1]),
cast("_InternalEntityType[Any]", cloned.path[0]),
):
raise sa_exc.ArgumentError(
f'Attribute "{cloned.path[1]}" does not link '
f'from element "{parent.path[-1]}".'
)
cloned.path = PathRegistry.coerce(parent.path[0:-1] + cloned.path[:])
cloned.path = PathRegistry.coerce(path[0:-1] + cloned.path[:])
return cloned
@@ -1789,7 +1841,7 @@ class _LoadElement(
replacement.local_opts = replacement.local_opts.union(
existing.local_opts
)
replacement._extra_criteria += replacement._extra_criteria
replacement._extra_criteria += existing._extra_criteria
return replacement
elif replacement.path.is_token:
# use 'last one wins' logic for wildcard options. this is also
@@ -1831,7 +1883,9 @@ class _AttributeStrategyLoad(_LoadElement):
is_class_strategy = False
is_token_strategy = False
def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
def _init_path(
self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
):
assert attr is not None
self._of_type = None
self._path_with_polymorphic_path = None
@@ -1877,7 +1931,11 @@ class _AttributeStrategyLoad(_LoadElement):
# from an attribute. This appears to have been an artifact of how
# _UnboundLoad / Load interacted together, which was opaque and
# poorly defined.
self._extra_criteria = attr._extra_criteria
if extra_criteria:
assert not attr._extra_criteria
self._extra_criteria = extra_criteria
else:
self._extra_criteria = attr._extra_criteria
if getattr(attr, "_of_type", None):
ac = attr._of_type
@@ -2070,7 +2128,9 @@ class _TokenStrategyLoad(_LoadElement):
is_class_strategy = False
is_token_strategy = True
def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
def _init_path(
self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
):
# assert isinstance(attr, str) or attr is None
if attr is not None:
default_token = attr.endswith(_DEFAULT_TOKEN)
@@ -2156,7 +2216,9 @@ class _ClassStrategyLoad(_LoadElement):
__visit_name__ = "class_strategy_load_element"
def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
def _init_path(
self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
):
return path
def _prepare_for_compile_state(
@@ -2447,7 +2509,9 @@ def _raise_for_does_not_link(path, attrname, parent_entity):
(
" Did you mean to use "
f'"{path[-2]}'
f'.of_type({parent_entity_str})"?'
f'.of_type({parent_entity_str})" or "loadopt.options('
f"selectin_polymorphic({path[-2].mapper.class_.__name__}, "
f'[{parent_entity_str}]), ...)" ?'
if not path_is_of_type
and not path[-1].is_aliased_class
and orm_util._entity_corresponds_to(