codekingpro/portable-devtools
115k
1# ext/associationproxy.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
8"""Contain the ``AssociationProxy`` class.
9
10The ``AssociationProxy`` is a Python property object which provides
11transparent proxied access to the endpoint of an association object.
12
13See the example ``examples/association/proxied_association.py``.
14
15"""
16from __future__ import annotations
17
18import operator
19import typing
20from typing import AbstractSet
21from typing import Any
22from typing import Callable
23from typing import cast
24from typing import Collection
25from typing import Dict
26from typing import Generic
27from typing import ItemsView
28from typing import Iterable
29from typing import Iterator
30from typing import KeysView
31from typing import List
32from typing import Mapping
33from typing import MutableMapping
34from typing import MutableSequence
35from typing import MutableSet
36from typing import NoReturn
37from typing import Optional
38from typing import overload
39from typing import Set
40from typing import Tuple
41from typing import Type
42from typing import TypeVar
43from typing import Union
44from typing import ValuesView
45
46from .. import ColumnElement
47from .. import exc
48from .. import inspect
49from .. import orm
50from .. import util
51from ..orm import collections
52from ..orm import InspectionAttrExtensionType
53from ..orm import interfaces
54from ..orm import ORMDescriptor
55from ..orm.base import SQLORMOperations
56from ..orm.interfaces import _AttributeOptions
57from ..orm.interfaces import _DCAttributeOptions
58from ..orm.interfaces import _DEFAULT_ATTRIBUTE_OPTIONS
59from ..sql import operators
60from ..sql import or_
61from ..sql.base import _NoArg
62from ..util.typing import Literal
63from ..util.typing import Protocol
64from ..util.typing import Self
65from ..util.typing import SupportsIndex
66from ..util.typing import SupportsKeysAndGetItem
67
68if typing.TYPE_CHECKING:
69 from ..orm.interfaces import MapperProperty
70 from ..orm.interfaces import PropComparator
71 from ..orm.mapper import Mapper
72 from ..sql._typing import _ColumnExpressionArgument
73 from ..sql._typing import _InfoType
74
75
76_T = TypeVar("_T", bound=Any)
77_T_co = TypeVar("_T_co", bound=Any, covariant=True)
78_T_con = TypeVar("_T_con", bound=Any, contravariant=True)
79_S = TypeVar("_S", bound=Any)
80_KT = TypeVar("_KT", bound=Any)
81_VT = TypeVar("_VT", bound=Any)
82
83
84def association_proxy(
85 target_collection: str,
86 attr: str,
87 *,
88 creator: Optional[_CreatorProtocol] = None,
89 getset_factory: Optional[_GetSetFactoryProtocol] = None,
90 proxy_factory: Optional[_ProxyFactoryProtocol] = None,
91 proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
92 info: Optional[_InfoType] = None,
93 cascade_scalar_deletes: bool = False,
94 create_on_none_assignment: bool = False,
95 init: Union[_NoArg, bool] = _NoArg.NO_ARG,
96 repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
97 default: Optional[Any] = _NoArg.NO_ARG,
98 default_factory: Union[_NoArg, Callable[[], _T]] = _NoArg.NO_ARG,
99 compare: Union[_NoArg, bool] = _NoArg.NO_ARG,
100 kw_only: Union[_NoArg, bool] = _NoArg.NO_ARG,
101 hash: Union[_NoArg, bool, None] = _NoArg.NO_ARG, # noqa: A002
102 dataclass_metadata: Union[_NoArg, Mapping[Any, Any], None] = _NoArg.NO_ARG,
103) -> AssociationProxy[Any]:
104 r"""Return a Python property implementing a view of a target
105 attribute which references an attribute on members of the
106 target.
107
108 The returned value is an instance of :class:`.AssociationProxy`.
109
110 Implements a Python property representing a relationship as a collection
111 of simpler values, or a scalar value. The proxied property will mimic
112 the collection type of the target (list, dict or set), or, in the case of
113 a one to one relationship, a simple scalar value.
114
115 :param target_collection: Name of the attribute that is the immediate
116 target. This attribute is typically mapped by
117 :func:`~sqlalchemy.orm.relationship` to link to a target collection, but
118 can also be a many-to-one or non-scalar relationship.
119
120 :param attr: Attribute on the associated instance or instances that
121 are available on instances of the target object.
122
123 :param creator: optional.
124
125 Defines custom behavior when new items are added to the proxied
126 collection.
127
128 By default, adding new items to the collection will trigger a
129 construction of an instance of the target object, passing the given
130 item as a positional argument to the target constructor. For cases
131 where this isn't sufficient, :paramref:`.association_proxy.creator`
132 can supply a callable that will construct the object in the
133 appropriate way, given the item that was passed.
134
135 For list- and set- oriented collections, a single argument is
136 passed to the callable. For dictionary oriented collections, two
137 arguments are passed, corresponding to the key and value.
138
139 The :paramref:`.association_proxy.creator` callable is also invoked
140 for scalar (i.e. many-to-one, one-to-one) relationships. If the
141 current value of the target relationship attribute is ``None``, the
142 callable is used to construct a new object. If an object value already
143 exists, the given attribute value is populated onto that object.
144
145 .. seealso::
146
147 :ref:`associationproxy_creator`
148
149 :param cascade_scalar_deletes: when True, indicates that setting
150 the proxied value to ``None``, or deleting it via ``del``, should
151 also remove the source object. Only applies to scalar attributes.
152 Normally, removing the proxied target will not remove the proxy
153 source, as this object may have other state that is still to be
154 kept.
155
156 .. versionadded:: 1.3
157
158 .. seealso::
159
160 :ref:`cascade_scalar_deletes` - complete usage example
161
162 :param create_on_none_assignment: when True, indicates that setting
163 the proxied value to ``None`` should **create** the source object
164 if it does not exist, using the creator. Only applies to scalar
165 attributes. This is mutually exclusive
166 vs. the :paramref:`.association_proxy.cascade_scalar_deletes`.
167
168 .. versionadded:: 2.0.18
169
170 :param init: Specific to :ref:`orm_declarative_native_dataclasses`,
171 specifies if the mapped attribute should be part of the ``__init__()``
172 method as generated by the dataclass process.
173
174 .. versionadded:: 2.0.0b4
175
176 :param repr: Specific to :ref:`orm_declarative_native_dataclasses`,
177 specifies if the attribute established by this :class:`.AssociationProxy`
178 should be part of the ``__repr__()`` method as generated by the dataclass
179 process.
180
181 .. versionadded:: 2.0.0b4
182
183 :param default_factory: Specific to
184 :ref:`orm_declarative_native_dataclasses`, specifies a default-value
185 generation function that will take place as part of the ``__init__()``
186 method as generated by the dataclass process.
187
188 .. versionadded:: 2.0.0b4
189
190 :param compare: Specific to
191 :ref:`orm_declarative_native_dataclasses`, indicates if this field
192 should be included in comparison operations when generating the
193 ``__eq__()`` and ``__ne__()`` methods for the mapped class.
194
195 .. versionadded:: 2.0.0b4
196
197 :param kw_only: Specific to :ref:`orm_declarative_native_dataclasses`,
198 indicates if this field should be marked as keyword-only when generating
199 the ``__init__()`` method as generated by the dataclass process.
200
201 .. versionadded:: 2.0.0b4
202
203 :param hash: Specific to
204 :ref:`orm_declarative_native_dataclasses`, controls if this field
205 is included when generating the ``__hash__()`` method for the mapped
206 class.
207
208 .. versionadded:: 2.0.36
209
210 :param dataclass_metadata: Specific to
211 :ref:`orm_declarative_native_dataclasses`, supplies metadata
212 to be attached to the generated dataclass field.
213
214 .. versionadded:: 2.0.42
215
216 :param info: optional, will be assigned to
217 :attr:`.AssociationProxy.info` if present.
218
219
220 The following additional parameters involve injection of custom behaviors
221 within the :class:`.AssociationProxy` object and are for advanced use
222 only:
223
224 :param getset_factory: Optional. Proxied attribute access is
225 automatically handled by routines that get and set values based on
226 the `attr` argument for this proxy.
227
228 If you would like to customize this behavior, you may supply a
229 `getset_factory` callable that produces a tuple of `getter` and
230 `setter` functions. The factory is called with two arguments, the
231 abstract type of the underlying collection and this proxy instance.
232
233 :param proxy_factory: Optional. The type of collection to emulate is
234 determined by sniffing the target collection. If your collection
235 type can't be determined by duck typing or you'd like to use a
236 different collection implementation, you may supply a factory
237 function to produce those collections. Only applicable to
238 non-scalar relationships.
239
240 :param proxy_bulk_set: Optional, use with proxy_factory.
241
242
243 """
244 return AssociationProxy(
245 target_collection,
246 attr,
247 creator=creator,
248 getset_factory=getset_factory,
249 proxy_factory=proxy_factory,
250 proxy_bulk_set=proxy_bulk_set,
251 info=info,
252 cascade_scalar_deletes=cascade_scalar_deletes,
253 create_on_none_assignment=create_on_none_assignment,
254 attribute_options=_AttributeOptions(
255 init,
256 repr,
257 default,
258 default_factory,
259 compare,
260 kw_only,
261 hash,
262 dataclass_metadata,
263 ),
264 )
265
266
267class AssociationProxyExtensionType(InspectionAttrExtensionType):
268 ASSOCIATION_PROXY = "ASSOCIATION_PROXY"
269 """Symbol indicating an :class:`.InspectionAttr` that's
270 of type :class:`.AssociationProxy`.
271
272 Is assigned to the :attr:`.InspectionAttr.extension_type`
273 attribute.
274
275 """
276
277
278class _GetterProtocol(Protocol[_T_co]):
279 def __call__(self, instance: Any) -> _T_co: ...
280
281
282# mypy 0.990 we are no longer allowed to make this Protocol[_T_con]
283class _SetterProtocol(Protocol): ...
284
285
286class _PlainSetterProtocol(_SetterProtocol, Protocol[_T_con]):
287 def __call__(self, instance: Any, value: _T_con) -> None: ...
288
289
290class _DictSetterProtocol(_SetterProtocol, Protocol[_T_con]):
291 def __call__(self, instance: Any, key: Any, value: _T_con) -> None: ...
292
293
294# mypy 0.990 we are no longer allowed to make this Protocol[_T_con]
295class _CreatorProtocol(Protocol): ...
296
297
298class _PlainCreatorProtocol(_CreatorProtocol, Protocol[_T_con]):
299 def __call__(self, value: _T_con) -> Any: ...
300
301
302class _KeyCreatorProtocol(_CreatorProtocol, Protocol[_T_con]):
303 def __call__(self, key: Any, value: Optional[_T_con]) -> Any: ...
304
305
306class _LazyCollectionProtocol(Protocol[_T]):
307 def __call__(
308 self,
309 ) -> Union[
310 MutableSet[_T], MutableMapping[Any, _T], MutableSequence[_T]
311 ]: ...
312
313
314class _GetSetFactoryProtocol(Protocol):
315 def __call__(
316 self,
317 collection_class: Optional[Type[Any]],
318 assoc_instance: AssociationProxyInstance[Any],
319 ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]: ...
320
321
322class _ProxyFactoryProtocol(Protocol):
323 def __call__(
324 self,
325 lazy_collection: _LazyCollectionProtocol[Any],
326 creator: _CreatorProtocol,
327 value_attr: str,
328 parent: AssociationProxyInstance[Any],
329 ) -> Any: ...
330
331
332class _ProxyBulkSetProtocol(Protocol):
333 def __call__(
334 self, proxy: _AssociationCollection[Any], collection: Iterable[Any]
335 ) -> None: ...
336
337
338class _AssociationProxyProtocol(Protocol[_T]):
339 """describes the interface of :class:`.AssociationProxy`
340 without including descriptor methods in the interface."""
341
342 creator: Optional[_CreatorProtocol]
343 key: str
344 target_collection: str
345 value_attr: str
346 cascade_scalar_deletes: bool
347 create_on_none_assignment: bool
348 getset_factory: Optional[_GetSetFactoryProtocol]
349 proxy_factory: Optional[_ProxyFactoryProtocol]
350 proxy_bulk_set: Optional[_ProxyBulkSetProtocol]
351
352 @util.ro_memoized_property
353 def info(self) -> _InfoType: ...
354
355 def for_class(
356 self, class_: Type[Any], obj: Optional[object] = None
357 ) -> AssociationProxyInstance[_T]: ...
358
359 def _default_getset(
360 self, collection_class: Any
361 ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]: ...
362
363
364class AssociationProxy(
365 interfaces.InspectionAttrInfo,
366 ORMDescriptor[_T],
367 _DCAttributeOptions,
368 _AssociationProxyProtocol[_T],
369):
370 """A descriptor that presents a read/write view of an object attribute."""
371
372 is_attribute = True
373 extension_type = AssociationProxyExtensionType.ASSOCIATION_PROXY
374
375 def __init__(
376 self,
377 target_collection: str,
378 attr: str,
379 *,
380 creator: Optional[_CreatorProtocol] = None,
381 getset_factory: Optional[_GetSetFactoryProtocol] = None,
382 proxy_factory: Optional[_ProxyFactoryProtocol] = None,
383 proxy_bulk_set: Optional[_ProxyBulkSetProtocol] = None,
384 info: Optional[_InfoType] = None,
385 cascade_scalar_deletes: bool = False,
386 create_on_none_assignment: bool = False,
387 attribute_options: Optional[_AttributeOptions] = None,
388 ):
389 """Construct a new :class:`.AssociationProxy`.
390
391 The :class:`.AssociationProxy` object is typically constructed using
392 the :func:`.association_proxy` constructor function. See the
393 description of :func:`.association_proxy` for a description of all
394 parameters.
395
396
397 """
398 self.target_collection = target_collection
399 self.value_attr = attr
400 self.creator = creator
401 self.getset_factory = getset_factory
402 self.proxy_factory = proxy_factory
403 self.proxy_bulk_set = proxy_bulk_set
404
405 if cascade_scalar_deletes and create_on_none_assignment:
406 raise exc.ArgumentError(
407 "The cascade_scalar_deletes and create_on_none_assignment "
408 "parameters are mutually exclusive."
409 )
410 self.cascade_scalar_deletes = cascade_scalar_deletes
411 self.create_on_none_assignment = create_on_none_assignment
412
413 self.key = "_%s_%s_%s" % (
414 type(self).__name__,
415 target_collection,
416 id(self),
417 )
418 if info:
419 self.info = info # type: ignore
420
421 if (
422 attribute_options
423 and attribute_options != _DEFAULT_ATTRIBUTE_OPTIONS
424 ):
425 self._has_dataclass_arguments = True
426 self._attribute_options = attribute_options
427 else:
428 self._has_dataclass_arguments = False
429 self._attribute_options = _DEFAULT_ATTRIBUTE_OPTIONS
430
431 @overload
432 def __get__(
433 self, instance: Literal[None], owner: Literal[None]
434 ) -> Self: ...
435
436 @overload
437 def __get__(
438 self, instance: Literal[None], owner: Any
439 ) -> AssociationProxyInstance[_T]: ...
440
441 @overload
442 def __get__(self, instance: object, owner: Any) -> _T: ...
443
444 def __get__(
445 self, instance: object, owner: Any
446 ) -> Union[AssociationProxyInstance[_T], _T, AssociationProxy[_T]]:
447 if owner is None:
448 return self
449 inst = self._as_instance(owner, instance)
450 if inst:
451 return inst.get(instance)
452
453 assert instance is None
454
455 return self
456
457 def __set__(self, instance: object, values: _T) -> None:
458 class_ = type(instance)
459 self._as_instance(class_, instance).set(instance, values)
460
461 def __delete__(self, instance: object) -> None:
462 class_ = type(instance)
463 self._as_instance(class_, instance).delete(instance)
464
465 def for_class(
466 self, class_: Type[Any], obj: Optional[object] = None
467 ) -> AssociationProxyInstance[_T]:
468 r"""Return the internal state local to a specific mapped class.
469
470 E.g., given a class ``User``::
471
472 class User(Base):
473 # ...
474
475 keywords = association_proxy("kws", "keyword")
476
477 If we access this :class:`.AssociationProxy` from
478 :attr:`_orm.Mapper.all_orm_descriptors`, and we want to view the
479 target class for this proxy as mapped by ``User``::
480
481 inspect(User).all_orm_descriptors["keywords"].for_class(User).target_class
482
483 This returns an instance of :class:`.AssociationProxyInstance` that
484 is specific to the ``User`` class. The :class:`.AssociationProxy`
485 object remains agnostic of its parent class.
486
487 :param class\_: the class that we are returning state for.
488
489 :param obj: optional, an instance of the class that is required
490 if the attribute refers to a polymorphic target, e.g. where we have
491 to look at the type of the actual destination object to get the
492 complete path.
493
494 .. versionadded:: 1.3 - :class:`.AssociationProxy` no longer stores
495 any state specific to a particular parent class; the state is now
496 stored in per-class :class:`.AssociationProxyInstance` objects.
497
498
499 """
500 return self._as_instance(class_, obj)
501
502 def _as_instance(
503 self, class_: Any, obj: Any
504 ) -> AssociationProxyInstance[_T]:
505 try:
506 inst = class_.__dict__[self.key + "_inst"]
507 except KeyError:
508 inst = None
509
510 # avoid exception context
511 if inst is None:
512 owner = self._calc_owner(class_)
513 if owner is not None:
514 inst = AssociationProxyInstance.for_proxy(self, owner, obj)
515 setattr(class_, self.key + "_inst", inst)
516 else:
517 inst = None
518
519 if inst is not None and not inst._is_canonical:
520 # the AssociationProxyInstance can't be generalized
521 # since the proxied attribute is not on the targeted
522 # class, only on subclasses of it, which might be
523 # different. only return for the specific
524 # object's current value
525 return inst._non_canonical_get_for_object(obj) # type: ignore
526 else:
527 return inst # type: ignore # TODO
528
529 def _calc_owner(self, target_cls: Any) -> Any:
530 # we might be getting invoked for a subclass
531 # that is not mapped yet, in some declarative situations.
532 # save until we are mapped
533 try:
534 insp = inspect(target_cls)
535 except exc.NoInspectionAvailable:
536 # can't find a mapper, don't set owner. if we are a not-yet-mapped
537 # subclass, we can also scan through __mro__ to find a mapped
538 # class, but instead just wait for us to be called again against a
539 # mapped class normally.
540 return None
541 else:
542 return insp.mapper.class_manager.class_
543
544 def _default_getset(
545 self, collection_class: Any
546 ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
547 attr = self.value_attr
548 _getter = operator.attrgetter(attr)
549
550 def getter(instance: Any) -> Optional[Any]:
551 return _getter(instance) if instance is not None else None
552
553 if collection_class is dict:
554
555 def dict_setter(instance: Any, k: Any, value: Any) -> None:
556 setattr(instance, attr, value)
557
558 return getter, dict_setter
559
560 else:
561
562 def plain_setter(o: Any, v: Any) -> None:
563 setattr(o, attr, v)
564
565 return getter, plain_setter
566
567 def __repr__(self) -> str:
568 return "AssociationProxy(%r, %r)" % (
569 self.target_collection,
570 self.value_attr,
571 )
572
573
574# the pep-673 Self type does not work in Mypy for a "hybrid"
575# style method that returns type or Self, so for one specific case
576# we still need to use the pre-pep-673 workaround.
577_Self = TypeVar("_Self", bound="AssociationProxyInstance[Any]")
578
579
580class AssociationProxyInstance(SQLORMOperations[_T]):
581 """A per-class object that serves class- and object-specific results.
582
583 This is used by :class:`.AssociationProxy` when it is invoked
584 in terms of a specific class or instance of a class, i.e. when it is
585 used as a regular Python descriptor.
586
587 When referring to the :class:`.AssociationProxy` as a normal Python
588 descriptor, the :class:`.AssociationProxyInstance` is the object that
589 actually serves the information. Under normal circumstances, its presence
590 is transparent::
591
592 >>> User.keywords.scalar
593 False
594
595 In the special case that the :class:`.AssociationProxy` object is being
596 accessed directly, in order to get an explicit handle to the
597 :class:`.AssociationProxyInstance`, use the
598 :meth:`.AssociationProxy.for_class` method::
599
600 proxy_state = inspect(User).all_orm_descriptors["keywords"].for_class(User)
601
602 # view if proxy object is scalar or not
603 >>> proxy_state.scalar
604 False
605
606 .. versionadded:: 1.3
607
608 """ # noqa
609
610 collection_class: Optional[Type[Any]]
611 parent: _AssociationProxyProtocol[_T]
612
613 def __init__(
614 self,
615 parent: _AssociationProxyProtocol[_T],
616 owning_class: Type[Any],
617 target_class: Type[Any],
618 value_attr: str,
619 ):
620 self.parent = parent
621 self.key = parent.key
622 self.owning_class = owning_class
623 self.target_collection = parent.target_collection
624 self.collection_class = None
625 self.target_class = target_class
626 self.value_attr = value_attr
627
628 target_class: Type[Any]
629 """The intermediary class handled by this
630 :class:`.AssociationProxyInstance`.
631
632 Intercepted append/set/assignment events will result
633 in the generation of new instances of this class.
634
635 """
636
637 @classmethod
638 def for_proxy(
639 cls,
640 parent: AssociationProxy[_T],
641 owning_class: Type[Any],
642 parent_instance: Any,
643 ) -> AssociationProxyInstance[_T]:
644 target_collection = parent.target_collection
645 value_attr = parent.value_attr
646 prop = cast(
647 "orm.RelationshipProperty[_T]",
648 orm.class_mapper(owning_class).get_property(target_collection),
649 )
650
651 # this was never asserted before but this should be made clear.
652 if not isinstance(prop, orm.RelationshipProperty):
653 raise NotImplementedError(
654 "association proxy to a non-relationship "
655 "intermediary is not supported"
656 ) from None
657
658 target_class = prop.mapper.class_
659
660 try:
661 target_assoc = cast(
662 "AssociationProxyInstance[_T]",
663 cls._cls_unwrap_target_assoc_proxy(target_class, value_attr),
664 )
665 except AttributeError:
666 # the proxied attribute doesn't exist on the target class;
667 # return an "ambiguous" instance that will work on a per-object
668 # basis
669 return AmbiguousAssociationProxyInstance(
670 parent, owning_class, target_class, value_attr
671 )
672 except Exception as err:
673 raise exc.InvalidRequestError(
674 f"Association proxy received an unexpected error when "
675 f"trying to retrieve attribute "
676 f'"{target_class.__name__}.{parent.value_attr}" from '
677 f'class "{target_class.__name__}": {err}'
678 ) from err
679 else:
680 return cls._construct_for_assoc(
681 target_assoc, parent, owning_class, target_class, value_attr
682 )
683
684 @classmethod
685 def _construct_for_assoc(
686 cls,
687 target_assoc: Optional[AssociationProxyInstance[_T]],
688 parent: _AssociationProxyProtocol[_T],
689 owning_class: Type[Any],
690 target_class: Type[Any],
691 value_attr: str,
692 ) -> AssociationProxyInstance[_T]:
693 if target_assoc is not None:
694 return ObjectAssociationProxyInstance(
695 parent, owning_class, target_class, value_attr
696 )
697
698 attr = getattr(target_class, value_attr)
699 if not hasattr(attr, "_is_internal_proxy"):
700 return AmbiguousAssociationProxyInstance(
701 parent, owning_class, target_class, value_attr
702 )
703 is_object = attr._impl_uses_objects
704 if is_object:
705 return ObjectAssociationProxyInstance(
706 parent, owning_class, target_class, value_attr
707 )
708 else:
709 return ColumnAssociationProxyInstance(
710 parent, owning_class, target_class, value_attr
711 )
712
713 def _get_property(self) -> MapperProperty[Any]:
714 return orm.class_mapper(self.owning_class).get_property(
715 self.target_collection
716 )
717
718 @property
719 def _comparator(self) -> PropComparator[Any]:
720 return getattr( # type: ignore
721 self.owning_class, self.target_collection
722 ).comparator
723
724 def __clause_element__(self) -> NoReturn:
725 raise NotImplementedError(
726 "The association proxy can't be used as a plain column "
727 "expression; it only works inside of a comparison expression"
728 )
729
730 @classmethod
731 def _cls_unwrap_target_assoc_proxy(
732 cls, target_class: Any, value_attr: str
733 ) -> Optional[AssociationProxyInstance[_T]]:
734 attr = getattr(target_class, value_attr)
735 assert not isinstance(attr, AssociationProxy)
736 if isinstance(attr, AssociationProxyInstance):
737 return attr
738 return None
739
740 @util.memoized_property
741 def _unwrap_target_assoc_proxy(
742 self,
743 ) -> Optional[AssociationProxyInstance[_T]]:
744 return self._cls_unwrap_target_assoc_proxy(
745 self.target_class, self.value_attr
746 )
747
748 @property
749 def remote_attr(self) -> SQLORMOperations[_T]:
750 """The 'remote' class attribute referenced by this
751 :class:`.AssociationProxyInstance`.
752
753 .. seealso::
754
755 :attr:`.AssociationProxyInstance.attr`
756
757 :attr:`.AssociationProxyInstance.local_attr`
758
759 """
760 return cast(
761 "SQLORMOperations[_T]", getattr(self.target_class, self.value_attr)
762 )
763
764 @property
765 def local_attr(self) -> SQLORMOperations[Any]:
766 """The 'local' class attribute referenced by this
767 :class:`.AssociationProxyInstance`.
768
769 .. seealso::
770
771 :attr:`.AssociationProxyInstance.attr`
772
773 :attr:`.AssociationProxyInstance.remote_attr`
774
775 """
776 return cast(
777 "SQLORMOperations[Any]",
778 getattr(self.owning_class, self.target_collection),
779 )
780
781 @property
782 def attr(self) -> Tuple[SQLORMOperations[Any], SQLORMOperations[_T]]:
783 """Return a tuple of ``(local_attr, remote_attr)``.
784
785 This attribute was originally intended to facilitate using the
786 :meth:`_query.Query.join` method to join across the two relationships
787 at once, however this makes use of a deprecated calling style.
788
789 To use :meth:`_sql.select.join` or :meth:`_orm.Query.join` with
790 an association proxy, the current method is to make use of the
791 :attr:`.AssociationProxyInstance.local_attr` and
792 :attr:`.AssociationProxyInstance.remote_attr` attributes separately::
793
794 stmt = (
795 select(Parent)
796 .join(Parent.proxied.local_attr)
797 .join(Parent.proxied.remote_attr)
798 )
799
800 A future release may seek to provide a more succinct join pattern
801 for association proxy attributes.
802
803 .. seealso::
804
805 :attr:`.AssociationProxyInstance.local_attr`
806
807 :attr:`.AssociationProxyInstance.remote_attr`
808
809 """
810 return (self.local_attr, self.remote_attr)
811
812 @util.memoized_property
813 def scalar(self) -> bool:
814 """Return ``True`` if this :class:`.AssociationProxyInstance`
815 proxies a scalar relationship on the local side."""
816
817 scalar = not self._get_property().uselist
818 if scalar:
819 self._initialize_scalar_accessors()
820 return scalar
821
822 @util.memoized_property
823 def _value_is_scalar(self) -> bool:
824 return (
825 not self._get_property()
826 .mapper.get_property(self.value_attr)
827 .uselist
828 )
829
830 @property
831 def _target_is_object(self) -> bool:
832 raise NotImplementedError()
833
834 _scalar_get: _GetterProtocol[_T]
835 _scalar_set: _PlainSetterProtocol[_T]
836
837 def _initialize_scalar_accessors(self) -> None:
838 if self.parent.getset_factory:
839 get, set_ = self.parent.getset_factory(None, self)
840 else:
841 get, set_ = self.parent._default_getset(None)
842 self._scalar_get, self._scalar_set = get, cast(
843 "_PlainSetterProtocol[_T]", set_
844 )
845
846 def _default_getset(
847 self, collection_class: Any
848 ) -> Tuple[_GetterProtocol[Any], _SetterProtocol]:
849 attr = self.value_attr
850 _getter = operator.attrgetter(attr)
851
852 def getter(instance: Any) -> Optional[_T]:
853 return _getter(instance) if instance is not None else None
854
855 if collection_class is dict:
856
857 def dict_setter(instance: Any, k: Any, value: _T) -> None:
858 setattr(instance, attr, value)
859
860 return getter, dict_setter
861 else:
862
863 def plain_setter(o: Any, v: _T) -> None:
864 setattr(o, attr, v)
865
866 return getter, plain_setter
867
868 @util.ro_non_memoized_property
869 def info(self) -> _InfoType:
870 return self.parent.info
871
872 @overload
873 def get(self: _Self, obj: Literal[None]) -> _Self: ...
874
875 @overload
876 def get(self, obj: Any) -> _T: ...
877
878 def get(
879 self, obj: Any
880 ) -> Union[Optional[_T], AssociationProxyInstance[_T]]:
881 if obj is None:
882 return self
883
884 proxy: _T
885
886 if self.scalar:
887 target = getattr(obj, self.target_collection)
888 return self._scalar_get(target)
889 else:
890 try:
891 # If the owning instance is reborn (orm session resurrect,
892 # etc.), refresh the proxy cache.
893 creator_id, self_id, proxy = cast(
894 "Tuple[int, int, _T]", getattr(obj, self.key)
895 )
896 except AttributeError:
897 pass
898 else:
899 if id(obj) == creator_id and id(self) == self_id:
900 assert self.collection_class is not None
901 return proxy
902
903 self.collection_class, proxy = self._new(
904 _lazy_collection(obj, self.target_collection)
905 )
906 setattr(obj, self.key, (id(obj), id(self), proxy))
907 return proxy
908
909 def set(self, obj: Any, values: _T) -> None:
910 if self.scalar:
911 creator = cast(
912 "_PlainCreatorProtocol[_T]",
913 (
914 self.parent.creator
915 if self.parent.creator
916 else self.target_class
917 ),
918 )
919 target = getattr(obj, self.target_collection)
920 if target is None:
921 if (
922 values is None
923 and not self.parent.create_on_none_assignment
924 ):
925 return
926 setattr(obj, self.target_collection, creator(values))
927 else:
928 self._scalar_set(target, values)
929 if values is None and self.parent.cascade_scalar_deletes:
930 setattr(obj, self.target_collection, None)
931 else:
932 proxy = self.get(obj)
933 assert self.collection_class is not None
934 if proxy is not values:
935 proxy._bulk_replace(self, values)
936
937 def delete(self, obj: Any) -> None:
938 if self.owning_class is None:
939 self._calc_owner(obj, None)
940
941 if self.scalar:
942 target = getattr(obj, self.target_collection)
943 if target is not None:
944 delattr(target, self.value_attr)
945 delattr(obj, self.target_collection)
946
947 def _new(
948 self, lazy_collection: _LazyCollectionProtocol[_T]
949 ) -> Tuple[Type[Any], _T]:
950 creator = (
951 self.parent.creator
952 if self.parent.creator is not None
953 else cast("_CreatorProtocol", self.target_class)
954 )
955 collection_class = util.duck_type_collection(lazy_collection())
956
957 if collection_class is None:
958 raise exc.InvalidRequestError(
959 f"lazy collection factory did not return a "
960 f"valid collection type, got {collection_class}"
961 )
962 if self.parent.proxy_factory:
963 return (
964 collection_class,
965 self.parent.proxy_factory(
966 lazy_collection, creator, self.value_attr, self
967 ),
968 )
969
970 if self.parent.getset_factory:
971 getter, setter = self.parent.getset_factory(collection_class, self)
972 else:
973 getter, setter = self.parent._default_getset(collection_class)
974
975 if collection_class is list:
976 return (
977 collection_class,
978 cast(
979 _T,
980 _AssociationList(
981 lazy_collection, creator, getter, setter, self
982 ),
983 ),
984 )
985 elif collection_class is dict:
986 return (
987 collection_class,
988 cast(
989 _T,
990 _AssociationDict(
991 lazy_collection, creator, getter, setter, self
992 ),
993 ),
994 )
995 elif collection_class is set:
996 return (
997 collection_class,
998 cast(
999 _T,
1000 _AssociationSet(
1001 lazy_collection, creator, getter, setter, self
1002 ),
1003 ),
1004 )
1005 else:
1006 raise exc.ArgumentError(
1007 "could not guess which interface to use for "
1008 'collection_class "%s" backing "%s"; specify a '
1009 "proxy_factory and proxy_bulk_set manually"
1010 % (self.collection_class, self.target_collection)
1011 )
1012
1013 def _set(
1014 self, proxy: _AssociationCollection[Any], values: Iterable[Any]
1015 ) -> None:
1016 if self.parent.proxy_bulk_set:
1017 self.parent.proxy_bulk_set(proxy, values)
1018 elif self.collection_class is list:
1019 cast("_AssociationList[Any]", proxy).extend(values)
1020 elif self.collection_class is dict:
1021 cast("_AssociationDict[Any, Any]", proxy).update(values)
1022 elif self.collection_class is set:
1023 cast("_AssociationSet[Any]", proxy).update(values)
1024 else:
1025 raise exc.ArgumentError(
1026 "no proxy_bulk_set supplied for custom "
1027 "collection_class implementation"
1028 )
1029
1030 def _inflate(self, proxy: _AssociationCollection[Any]) -> None:
1031 creator = (
1032 self.parent.creator
1033 and self.parent.creator
1034 or cast(_CreatorProtocol, self.target_class)
1035 )
1036
1037 if self.parent.getset_factory:
1038 getter, setter = self.parent.getset_factory(
1039 self.collection_class, self
1040 )
1041 else:
1042 getter, setter = self.parent._default_getset(self.collection_class)
1043
1044 proxy.creator = creator
1045 proxy.getter = getter
1046 proxy.setter = setter
1047
1048 def _criterion_exists(
1049 self,
1050 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1051 **kwargs: Any,
1052 ) -> ColumnElement[bool]:
1053 is_has = kwargs.pop("is_has", None)
1054
1055 target_assoc = self._unwrap_target_assoc_proxy
1056 if target_assoc is not None:
1057 inner = target_assoc._criterion_exists(
1058 criterion=criterion, **kwargs
1059 )
1060 return self._comparator._criterion_exists(inner)
1061
1062 if self._target_is_object:
1063 attr = getattr(self.target_class, self.value_attr)
1064 value_expr = attr.comparator._criterion_exists(criterion, **kwargs)
1065 else:
1066 if kwargs:
1067 raise exc.ArgumentError(
1068 "Can't apply keyword arguments to column-targeted "
1069 "association proxy; use =="
1070 )
1071 elif is_has and criterion is not None:
1072 raise exc.ArgumentError(
1073 "Non-empty has() not allowed for "
1074 "column-targeted association proxy; use =="
1075 )
1076
1077 value_expr = criterion
1078
1079 return self._comparator._criterion_exists(value_expr)
1080
1081 def any(
1082 self,
1083 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1084 **kwargs: Any,
1085 ) -> ColumnElement[bool]:
1086 """Produce a proxied 'any' expression using EXISTS.
1087
1088 This expression will be a composed product
1089 using the :meth:`.Relationship.Comparator.any`
1090 and/or :meth:`.Relationship.Comparator.has`
1091 operators of the underlying proxied attributes.
1092
1093 """
1094 if self._unwrap_target_assoc_proxy is None and (
1095 self.scalar
1096 and (not self._target_is_object or self._value_is_scalar)
1097 ):
1098 raise exc.InvalidRequestError(
1099 "'any()' not implemented for scalar attributes. Use has()."
1100 )
1101 return self._criterion_exists(
1102 criterion=criterion, is_has=False, **kwargs
1103 )
1104
1105 def has(
1106 self,
1107 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1108 **kwargs: Any,
1109 ) -> ColumnElement[bool]:
1110 """Produce a proxied 'has' expression using EXISTS.
1111
1112 This expression will be a composed product
1113 using the :meth:`.Relationship.Comparator.any`
1114 and/or :meth:`.Relationship.Comparator.has`
1115 operators of the underlying proxied attributes.
1116
1117 """
1118 if self._unwrap_target_assoc_proxy is None and (
1119 not self.scalar
1120 or (self._target_is_object and not self._value_is_scalar)
1121 ):
1122 raise exc.InvalidRequestError(
1123 "'has()' not implemented for collections. Use any()."
1124 )
1125 return self._criterion_exists(
1126 criterion=criterion, is_has=True, **kwargs
1127 )
1128
1129 def __repr__(self) -> str:
1130 return "%s(%r)" % (self.__class__.__name__, self.parent)
1131
1132
1133class AmbiguousAssociationProxyInstance(AssociationProxyInstance[_T]):
1134 """an :class:`.AssociationProxyInstance` where we cannot determine
1135 the type of target object.
1136 """
1137
1138 _is_canonical = False
1139
1140 def _ambiguous(self) -> NoReturn:
1141 raise AttributeError(
1142 "Association proxy %s.%s refers to an attribute '%s' that is not "
1143 "directly mapped on class %s; therefore this operation cannot "
1144 "proceed since we don't know what type of object is referred "
1145 "towards"
1146 % (
1147 self.owning_class.__name__,
1148 self.target_collection,
1149 self.value_attr,
1150 self.target_class,
1151 )
1152 )
1153
1154 def get(self, obj: Any) -> Any:
1155 if obj is None:
1156 return self
1157 else:
1158 return super().get(obj)
1159
1160 def __eq__(self, obj: object) -> NoReturn:
1161 self._ambiguous()
1162
1163 def __ne__(self, obj: object) -> NoReturn:
1164 self._ambiguous()
1165
1166 def any(
1167 self,
1168 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1169 **kwargs: Any,
1170 ) -> NoReturn:
1171 self._ambiguous()
1172
1173 def has(
1174 self,
1175 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1176 **kwargs: Any,
1177 ) -> NoReturn:
1178 self._ambiguous()
1179
1180 @util.memoized_property
1181 def _lookup_cache(self) -> Dict[Type[Any], AssociationProxyInstance[_T]]:
1182 # mapping of <subclass>->AssociationProxyInstance.
1183 # e.g. proxy is A-> A.b -> B -> B.b_attr, but B.b_attr doesn't exist;
1184 # only B1(B) and B2(B) have "b_attr", keys in here would be B1, B2
1185 return {}
1186
1187 def _non_canonical_get_for_object(
1188 self, parent_instance: Any
1189 ) -> AssociationProxyInstance[_T]:
1190 if parent_instance is not None:
1191 actual_obj = getattr(parent_instance, self.target_collection)
1192 if actual_obj is not None:
1193 try:
1194 insp = inspect(actual_obj)
1195 except exc.NoInspectionAvailable:
1196 pass
1197 else:
1198 mapper = insp.mapper
1199 instance_class = mapper.class_
1200 if instance_class not in self._lookup_cache: