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