Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/relationships.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
8"""Heuristics related to join conditions as used in
9:func:`_orm.relationship`.
11Provides the :class:`.JoinCondition` object, which encapsulates
12SQL annotation and aliasing behavior focused on the `primaryjoin`
13and `secondaryjoin` aspects of :func:`_orm.relationship`.
15"""
17from __future__ import annotations
19import collections
20from collections import abc
21import dataclasses
22import inspect as _py_inspect
23import itertools
24import re
25import typing
26from typing import Any
27from typing import Callable
28from typing import cast
29from typing import Collection
30from typing import Dict
31from typing import FrozenSet
32from typing import Generic
33from typing import Iterable
34from typing import Iterator
35from typing import List
36from typing import Literal
37from typing import NamedTuple
38from typing import NoReturn
39from typing import Optional
40from typing import Sequence
41from typing import Set
42from typing import Tuple
43from typing import Type
44from typing import TYPE_CHECKING
45from typing import TypeVar
46from typing import Union
47import weakref
49from . import attributes
50from . import strategy_options
51from ._typing import insp_is_aliased_class
52from ._typing import is_has_collection_adapter
53from .base import _DeclarativeMapped
54from .base import _is_mapped_class
55from .base import class_mapper
56from .base import DynamicMapped
57from .base import LoaderCallableStatus
58from .base import PassiveFlag
59from .base import state_str
60from .base import WriteOnlyMapped
61from .interfaces import _AttributeOptions
62from .interfaces import _DataclassDefaultsDontSet
63from .interfaces import _IntrospectsAnnotations
64from .interfaces import MANYTOMANY
65from .interfaces import MANYTOONE
66from .interfaces import ONETOMANY
67from .interfaces import PropComparator
68from .interfaces import RelationshipDirection
69from .interfaces import StrategizedProperty
70from .util import CascadeOptions
71from .. import exc as sa_exc
72from .. import Exists
73from .. import log
74from .. import schema
75from .. import sql
76from .. import util
77from ..inspection import inspect
78from ..sql import coercions
79from ..sql import expression
80from ..sql import operators
81from ..sql import roles
82from ..sql import visitors
83from ..sql._typing import _ColumnExpressionArgument
84from ..sql._typing import _HasClauseElement
85from ..sql.annotation import _safe_annotate
86from ..sql.base import _NoArg
87from ..sql.elements import ColumnClause
88from ..sql.elements import ColumnElement
89from ..sql.util import _deep_annotate
90from ..sql.util import _deep_deannotate
91from ..sql.util import _shallow_annotate
92from ..sql.util import adapt_criterion_to_null
93from ..sql.util import ClauseAdapter
94from ..sql.util import join_condition
95from ..sql.util import selectables_overlap
96from ..sql.util import visit_binary_product
97from ..util.typing import de_optionalize_union_types
98from ..util.typing import resolve_name_to_real_class_name
100if typing.TYPE_CHECKING:
101 from ._typing import _EntityType
102 from ._typing import _ExternalEntityType
103 from ._typing import _IdentityKeyType
104 from ._typing import _InstanceDict
105 from ._typing import _InternalEntityType
106 from ._typing import _O
107 from ._typing import _RegistryType
108 from .base import Mapped
109 from .clsregistry import _class_resolver
110 from .clsregistry import _ModNS
111 from .decl_base import _DeclarativeMapperConfig
112 from .dependency import _DependencyProcessor
113 from .mapper import Mapper
114 from .query import Query
115 from .session import Session
116 from .state import InstanceState
117 from .strategies import _LazyLoader
118 from .util import AliasedClass
119 from .util import AliasedInsp
120 from ..sql._typing import _CoreAdapterProto
121 from ..sql._typing import _EquivalentColumnMap
122 from ..sql._typing import _InfoType
123 from ..sql.annotation import _AnnotationDict
124 from ..sql.annotation import SupportsAnnotations
125 from ..sql.elements import BinaryExpression
126 from ..sql.elements import BindParameter
127 from ..sql.elements import ClauseElement
128 from ..sql.schema import Table
129 from ..sql.selectable import FromClause
130 from ..util.typing import _AnnotationScanType
131 from ..util.typing import RODescriptorReference
133_T = TypeVar("_T", bound=Any)
134_T1 = TypeVar("_T1", bound=Any)
135_T2 = TypeVar("_T2", bound=Any)
137_PT = TypeVar("_PT", bound=Any)
139_PT2 = TypeVar("_PT2", bound=Any)
142_RelationshipArgumentType = Union[
143 str,
144 Type[_T],
145 Callable[[], Type[_T]],
146 "Mapper[_T]",
147 "AliasedClass[_T]",
148 Callable[[], "Mapper[_T]"],
149 Callable[[], "AliasedClass[_T]"],
150]
152_LazyLoadArgumentType = Literal[
153 "select",
154 "joined",
155 "selectin",
156 "subquery",
157 "raise",
158 "raise_on_sql",
159 "noload",
160 "immediate",
161 "write_only",
162 "dynamic",
163 True,
164 False,
165 None,
166]
169_RelationshipJoinConditionArgument = Union[
170 str, _ColumnExpressionArgument[bool]
171]
172_RelationshipSecondaryArgument = Union[
173 "FromClause", str, Callable[[], "FromClause"]
174]
175_ORMOrderByArgument = Union[
176 Literal[False],
177 str,
178 _ColumnExpressionArgument[Any],
179 Callable[[], _ColumnExpressionArgument[Any]],
180 Callable[[], Iterable[_ColumnExpressionArgument[Any]]],
181 Iterable[Union[str, _ColumnExpressionArgument[Any]]],
182]
183_RelationshipBackPopulatesArgument = Union[
184 str,
185 PropComparator[Any],
186 Callable[[], Union[str, PropComparator[Any]]],
187]
190ORMBackrefArgument = Union[str, Tuple[str, Dict[str, Any]]]
192_ORMColCollectionElement = Union[
193 ColumnClause[Any],
194 _HasClauseElement[Any],
195 roles.DMLColumnRole,
196 "Mapped[Any]",
197]
198_ORMColCollectionArgument = Union[
199 str,
200 Sequence[_ORMColCollectionElement],
201 Callable[[], Sequence[_ORMColCollectionElement]],
202 Callable[[], _ORMColCollectionElement],
203 _ORMColCollectionElement,
204]
207_CEA = TypeVar("_CEA", bound=_ColumnExpressionArgument[Any])
209_CE = TypeVar("_CE", bound="ColumnElement[Any]")
212_ColumnPairIterable = Iterable[Tuple[ColumnElement[Any], ColumnElement[Any]]]
214_ColumnPairs = Sequence[Tuple[ColumnElement[Any], ColumnElement[Any]]]
216_MutableColumnPairs = List[Tuple[ColumnElement[Any], ColumnElement[Any]]]
219def remote(expr: _CEA) -> _CEA:
220 """Annotate a portion of a primaryjoin expression
221 with a 'remote' annotation.
223 See the section :ref:`relationship_custom_foreign` for a
224 description of use.
226 .. seealso::
228 :ref:`relationship_custom_foreign`
230 :func:`.foreign`
232 """
233 return _annotate_columns( # type: ignore[return-value]
234 coercions.expect(roles.ColumnArgumentRole, expr), {"remote": True}
235 )
238def foreign(expr: _CEA) -> _CEA:
239 """Annotate a portion of a primaryjoin expression
240 with a 'foreign' annotation.
242 See the section :ref:`relationship_custom_foreign` for a
243 description of use.
245 .. seealso::
247 :ref:`relationship_custom_foreign`
249 :func:`.remote`
251 """
253 return _annotate_columns( # type: ignore[return-value]
254 coercions.expect(roles.ColumnArgumentRole, expr), {"foreign": True}
255 )
258@dataclasses.dataclass
259class _RelationshipArg(Generic[_T1, _T2]):
260 """stores a user-defined parameter value that must be resolved and
261 parsed later at mapper configuration time.
263 """
265 __slots__ = "name", "argument", "resolved"
266 name: str
267 argument: _T1
268 resolved: Optional[_T2]
270 def _is_populated(self) -> bool:
271 return self.argument is not None
273 def _resolve_against_registry(
274 self, clsregistry_resolver: Callable[[str, bool], _class_resolver]
275 ) -> None:
276 attr_value = self.argument
278 if isinstance(attr_value, str):
279 self.resolved = clsregistry_resolver(
280 attr_value, self.name == "secondary"
281 )()
282 elif callable(attr_value) and not _is_mapped_class(attr_value):
283 self.resolved = attr_value()
284 else:
285 self.resolved = attr_value
287 def effective_value(self) -> Any:
288 if self.resolved is not None:
289 return self.resolved
290 else:
291 return self.argument
294_RelationshipOrderByArg = Union[Literal[False], Tuple[ColumnElement[Any], ...]]
297@dataclasses.dataclass
298class _StringRelationshipArg(_RelationshipArg[_T1, _T2]):
299 def _resolve_against_registry(
300 self, clsregistry_resolver: Callable[[str, bool], _class_resolver]
301 ) -> None:
302 attr_value = self.argument
304 if callable(attr_value):
305 attr_value = attr_value()
307 if isinstance(attr_value, attributes.QueryableAttribute):
308 attr_value = attr_value.key # type: ignore[assignment]
310 self.resolved = attr_value
313class _RelationshipArgs(NamedTuple):
314 """stores user-passed parameters that are resolved at mapper configuration
315 time.
317 """
319 secondary: _RelationshipArg[
320 Optional[_RelationshipSecondaryArgument],
321 Optional[FromClause],
322 ]
323 primaryjoin: _RelationshipArg[
324 Optional[_RelationshipJoinConditionArgument],
325 Optional[ColumnElement[Any]],
326 ]
327 secondaryjoin: _RelationshipArg[
328 Optional[_RelationshipJoinConditionArgument],
329 Optional[ColumnElement[Any]],
330 ]
331 order_by: _RelationshipArg[_ORMOrderByArgument, _RelationshipOrderByArg]
332 foreign_keys: _RelationshipArg[
333 Optional[_ORMColCollectionArgument], Set[ColumnElement[Any]]
334 ]
335 remote_side: _RelationshipArg[
336 Optional[_ORMColCollectionArgument], Set[ColumnElement[Any]]
337 ]
338 back_populates: _StringRelationshipArg[
339 Optional[_RelationshipBackPopulatesArgument], str
340 ]
343@log.class_logger
344class RelationshipProperty(
345 _DataclassDefaultsDontSet,
346 _IntrospectsAnnotations,
347 StrategizedProperty[_T],
348 log.Identified,
349):
350 """Describes an object property that holds a single item or list
351 of items that correspond to a related database table.
353 Public constructor is the :func:`_orm.relationship` function.
355 .. seealso::
357 :ref:`relationship_config_toplevel`
359 """
361 strategy_wildcard_key = strategy_options._RELATIONSHIP_TOKEN
362 inherit_cache = True
363 """:meta private:"""
365 _links_to_entity = True
366 _is_relationship = True
368 _overlaps: Sequence[str]
370 _lazy_strategy: _LazyLoader
372 _persistence_only = dict(
373 passive_deletes=False,
374 passive_updates=True,
375 enable_typechecks=True,
376 active_history=False,
377 cascade_backrefs=False,
378 )
380 _dependency_processor: Optional[_DependencyProcessor] = None
382 primaryjoin: ColumnElement[bool]
383 secondaryjoin: Optional[ColumnElement[bool]]
384 secondary: Optional[FromClause]
385 _join_condition: _JoinCondition
386 order_by: _RelationshipOrderByArg
388 _user_defined_foreign_keys: Set[ColumnElement[Any]]
389 _calculated_foreign_keys: Set[ColumnElement[Any]]
391 remote_side: Set[ColumnElement[Any]]
392 local_columns: Set[ColumnElement[Any]]
394 synchronize_pairs: _ColumnPairs
395 secondary_synchronize_pairs: Optional[_ColumnPairs]
397 local_remote_pairs: _ColumnPairs
399 direction: RelationshipDirection
401 _init_args: _RelationshipArgs
402 _disable_dataclass_default_factory = True
404 def __init__(
405 self,
406 argument: Optional[_RelationshipArgumentType[_T]] = None,
407 secondary: Optional[_RelationshipSecondaryArgument] = None,
408 *,
409 uselist: Optional[bool] = None,
410 collection_class: Optional[
411 Union[Type[Collection[Any]], Callable[[], Collection[Any]]]
412 ] = None,
413 primaryjoin: Optional[_RelationshipJoinConditionArgument] = None,
414 secondaryjoin: Optional[_RelationshipJoinConditionArgument] = None,
415 back_populates: Optional[_RelationshipBackPopulatesArgument] = None,
416 order_by: _ORMOrderByArgument = False,
417 backref: Optional[ORMBackrefArgument] = None,
418 overlaps: Optional[str] = None,
419 post_update: bool = False,
420 cascade: str = "save-update, merge",
421 viewonly: bool = False,
422 attribute_options: Optional[_AttributeOptions] = None,
423 lazy: _LazyLoadArgumentType = "select",
424 passive_deletes: Union[Literal["all"], bool] = False,
425 passive_updates: bool = True,
426 active_history: bool = False,
427 enable_typechecks: bool = True,
428 foreign_keys: Optional[_ORMColCollectionArgument] = None,
429 remote_side: Optional[_ORMColCollectionArgument] = None,
430 join_depth: Optional[int] = None,
431 comparator_factory: Optional[
432 Type[RelationshipProperty.Comparator[Any]]
433 ] = None,
434 single_parent: bool = False,
435 innerjoin: bool = False,
436 distinct_target_key: Optional[bool] = None,
437 load_on_pending: bool = False,
438 query_class: Optional[Type[Query[Any]]] = None,
439 info: Optional[_InfoType] = None,
440 omit_join: Literal[None, False] = None,
441 sync_backref: Optional[bool] = None,
442 doc: Optional[str] = None,
443 bake_queries: Literal[True] = True,
444 cascade_backrefs: Literal[False] = False,
445 _local_remote_pairs: Optional[_ColumnPairs] = None,
446 _legacy_inactive_history_style: bool = False,
447 ):
448 super().__init__(attribute_options=attribute_options)
450 self.uselist = uselist
451 self.argument = argument
453 self._init_args = _RelationshipArgs(
454 _RelationshipArg("secondary", secondary, None),
455 _RelationshipArg("primaryjoin", primaryjoin, None),
456 _RelationshipArg("secondaryjoin", secondaryjoin, None),
457 _RelationshipArg("order_by", order_by, None),
458 _RelationshipArg("foreign_keys", foreign_keys, None),
459 _RelationshipArg("remote_side", remote_side, None),
460 _StringRelationshipArg("back_populates", back_populates, None),
461 )
463 if self._attribute_options.dataclasses_default not in (
464 _NoArg.NO_ARG,
465 None,
466 ):
467 raise sa_exc.ArgumentError(
468 "Only 'None' is accepted as dataclass "
469 "default for a relationship()"
470 )
472 self.post_update = post_update
473 self.viewonly = viewonly
474 if viewonly:
475 self._warn_for_persistence_only_flags(
476 passive_deletes=passive_deletes,
477 passive_updates=passive_updates,
478 enable_typechecks=enable_typechecks,
479 active_history=active_history,
480 cascade_backrefs=cascade_backrefs,
481 )
482 if viewonly and sync_backref:
483 raise sa_exc.ArgumentError(
484 "sync_backref and viewonly cannot both be True"
485 )
486 self.sync_backref = sync_backref
487 self.lazy = lazy
488 self.single_parent = single_parent
489 self.collection_class = collection_class
490 self.passive_deletes = passive_deletes
492 if cascade_backrefs:
493 raise sa_exc.ArgumentError(
494 "The 'cascade_backrefs' parameter passed to "
495 "relationship() may only be set to False."
496 )
498 self.passive_updates = passive_updates
499 self.enable_typechecks = enable_typechecks
500 self.query_class = query_class
501 self.innerjoin = innerjoin
502 self.distinct_target_key = distinct_target_key
503 self.doc = doc
504 self.active_history = active_history
505 self._legacy_inactive_history_style = _legacy_inactive_history_style
507 self.join_depth = join_depth
508 if omit_join:
509 util.warn(
510 "setting omit_join to True is not supported; selectin "
511 "loading of this relationship may not work correctly if this "
512 "flag is set explicitly. omit_join optimization is "
513 "automatically detected for conditions under which it is "
514 "supported."
515 )
517 self.omit_join = omit_join
518 self.local_remote_pairs = _local_remote_pairs or ()
519 self.load_on_pending = load_on_pending
520 self.comparator_factory = (
521 comparator_factory or RelationshipProperty.Comparator
522 )
523 util.set_creation_order(self)
525 if info is not None:
526 self.info.update(info)
528 self.strategy_key = (("lazy", self.lazy),)
530 self._reverse_property: Set[RelationshipProperty[Any]] = set()
532 if overlaps:
533 self._overlaps = set(re.split(r"\s*,\s*", overlaps)) # type: ignore[assignment] # noqa: E501
534 else:
535 self._overlaps = ()
537 self.cascade = cascade
539 if back_populates:
540 if backref:
541 raise sa_exc.ArgumentError(
542 "backref and back_populates keyword arguments "
543 "are mutually exclusive"
544 )
545 self.backref = None
546 else:
547 self.backref = backref
549 @property
550 def back_populates(self) -> str:
551 return self._init_args.back_populates.effective_value() # type: ignore[no-any-return] # noqa: E501
553 @back_populates.setter
554 def back_populates(self, value: str) -> None:
555 self._init_args.back_populates.argument = value
557 def _warn_for_persistence_only_flags(self, **kw: Any) -> None:
558 for k, v in kw.items():
559 if v != self._persistence_only[k]:
560 # we are warning here rather than warn deprecated as this is a
561 # configuration mistake, and Python shows regular warnings more
562 # aggressively than deprecation warnings by default. Unlike the
563 # case of setting viewonly with cascade, the settings being
564 # warned about here are not actively doing the wrong thing
565 # against viewonly=True, so it is not as urgent to have these
566 # raise an error.
567 util.warn(
568 "Setting %s on relationship() while also "
569 "setting viewonly=True does not make sense, as a "
570 "viewonly=True relationship does not perform persistence "
571 "operations. This configuration may raise an error "
572 "in a future release." % (k,)
573 )
575 def instrument_class(self, mapper: Mapper[Any]) -> None:
576 attributes._register_descriptor(
577 mapper.class_,
578 self.key,
579 comparator=self.comparator_factory(self, mapper),
580 parententity=mapper,
581 doc=self.doc,
582 )
584 class Comparator(util.MemoizedSlots, PropComparator[_PT]):
585 """Produce boolean, comparison, and other operators for
586 :class:`.RelationshipProperty` attributes.
588 See the documentation for :class:`.PropComparator` for a brief
589 overview of ORM level operator definition.
591 .. seealso::
593 :class:`.PropComparator`
595 :class:`.ColumnProperty.Comparator`
597 :class:`.ColumnOperators`
599 :ref:`types_operators`
601 :attr:`.TypeEngine.comparator_factory`
603 """
605 __slots__ = (
606 "entity",
607 "mapper",
608 "property",
609 "_of_type",
610 "_extra_criteria",
611 )
613 prop: RODescriptorReference[RelationshipProperty[_PT]]
614 _of_type: Optional[_EntityType[_PT]]
616 def __init__(
617 self,
618 prop: RelationshipProperty[_PT],
619 parentmapper: _InternalEntityType[Any],
620 adapt_to_entity: Optional[AliasedInsp[Any]] = None,
621 of_type: Optional[_EntityType[_PT]] = None,
622 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
623 ):
624 """Construction of :class:`.RelationshipProperty.Comparator`
625 is internal to the ORM's attribute mechanics.
627 """
628 self.prop = prop
629 self._parententity = parentmapper
630 self._adapt_to_entity = adapt_to_entity
631 if of_type:
632 self._of_type = of_type
633 else:
634 self._of_type = None
635 self._extra_criteria = extra_criteria
637 def adapt_to_entity(
638 self, adapt_to_entity: AliasedInsp[Any]
639 ) -> RelationshipProperty.Comparator[Any]:
640 return self.__class__(
641 self.prop,
642 self._parententity,
643 adapt_to_entity=adapt_to_entity,
644 of_type=self._of_type,
645 )
647 entity: _InternalEntityType[_PT]
648 """The target entity referred to by this
649 :class:`.RelationshipProperty.Comparator`.
651 This is either a :class:`_orm.Mapper` or :class:`.AliasedInsp`
652 object.
654 This is the "target" or "remote" side of the
655 :func:`_orm.relationship`.
657 """
659 mapper: Mapper[_PT]
660 """The target :class:`_orm.Mapper` referred to by this
661 :class:`.RelationshipProperty.Comparator`.
663 This is the "target" or "remote" side of the
664 :func:`_orm.relationship`.
666 """
668 def _memoized_attr_entity(self) -> _InternalEntityType[_PT]:
669 if self._of_type:
670 return inspect(self._of_type) # type: ignore[return-value]
671 else:
672 return self.prop.entity
674 def _memoized_attr_mapper(self) -> Mapper[_PT]:
675 return self.entity.mapper
677 def _source_selectable(self) -> FromClause:
678 if self._adapt_to_entity:
679 return self._adapt_to_entity.selectable
680 else:
681 return self.property.parent._with_polymorphic_selectable
683 def __clause_element__(self) -> ColumnElement[bool]:
684 adapt_from = self._source_selectable()
685 if self._of_type:
686 of_type_entity = inspect(self._of_type)
687 else:
688 of_type_entity = None
690 (
691 pj,
692 sj,
693 source,
694 dest,
695 secondary,
696 target_adapter,
697 ) = self.prop._create_joins(
698 source_selectable=adapt_from,
699 source_polymorphic=True,
700 of_type_entity=of_type_entity,
701 alias_secondary=True,
702 extra_criteria=self._extra_criteria,
703 )
704 if sj is not None:
705 return pj & sj
706 else:
707 return pj
709 def of_type(self, class_: _EntityType[Any]) -> PropComparator[_PT]:
710 r"""Redefine this object in terms of a polymorphic subclass.
712 See :meth:`.PropComparator.of_type` for an example.
715 """
716 return RelationshipProperty.Comparator(
717 self.prop,
718 self._parententity,
719 adapt_to_entity=self._adapt_to_entity,
720 of_type=class_,
721 extra_criteria=self._extra_criteria,
722 )
724 def and_(
725 self, *criteria: _ColumnExpressionArgument[bool]
726 ) -> PropComparator[Any]:
727 """Add AND criteria.
729 See :meth:`.PropComparator.and_` for an example.
731 .. versionadded:: 1.4
733 """
734 exprs = tuple(
735 coercions.expect(roles.WhereHavingRole, clause)
736 for clause in util.coerce_generator_arg(criteria)
737 )
739 return RelationshipProperty.Comparator(
740 self.prop,
741 self._parententity,
742 adapt_to_entity=self._adapt_to_entity,
743 of_type=self._of_type,
744 extra_criteria=self._extra_criteria + exprs,
745 )
747 def in_(self, other: Any) -> NoReturn:
748 """Produce an IN clause - this is not implemented
749 for :func:`_orm.relationship`-based attributes at this time.
751 """
752 raise NotImplementedError(
753 "in_() not yet supported for "
754 "relationships. For a simple "
755 "many-to-one, use in_() against "
756 "the set of foreign key values."
757 )
759 # https://github.com/python/mypy/issues/4266
760 __hash__ = None # type: ignore[assignment]
762 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
763 """Implement the ``==`` operator.
765 In a many-to-one context, such as:
767 .. sourcecode:: text
769 MyClass.some_prop == <some object>
771 this will typically produce a
772 clause such as:
774 .. sourcecode:: text
776 mytable.related_id == <some id>
778 Where ``<some id>`` is the primary key of the given
779 object.
781 The ``==`` operator provides partial functionality for non-
782 many-to-one comparisons:
784 * Comparisons against collections are not supported.
785 Use :meth:`~.RelationshipProperty.Comparator.contains`.
786 * Compared to a scalar one-to-many, will produce a
787 clause that compares the target columns in the parent to
788 the given target.
789 * Compared to a scalar many-to-many, an alias
790 of the association table will be rendered as
791 well, forming a natural join that is part of the
792 main body of the query. This will not work for
793 queries that go beyond simple AND conjunctions of
794 comparisons, such as those which use OR. Use
795 explicit joins, outerjoins, or
796 :meth:`~.RelationshipProperty.Comparator.has` for
797 more comprehensive non-many-to-one scalar
798 membership tests.
799 * Comparisons against ``None`` given in a one-to-many
800 or many-to-many context produce a NOT EXISTS clause.
802 """
803 if other is None or isinstance(other, expression.Null):
804 if self.property.direction in [ONETOMANY, MANYTOMANY]:
805 return ~self._criterion_exists()
806 else:
807 return self.property._optimized_compare(
808 None, adapt_source=self.adapter
809 )
810 elif self.property.uselist:
811 raise sa_exc.InvalidRequestError(
812 "Can't compare a collection to an object or collection; "
813 "use contains() to test for membership."
814 )
815 else:
816 return self.property._optimized_compare(
817 other, adapt_source=self.adapter
818 )
820 def _criterion_exists(
821 self,
822 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
823 **kwargs: Any,
824 ) -> Exists:
825 where_criteria = (
826 coercions.expect(roles.WhereHavingRole, criterion)
827 if criterion is not None
828 else None
829 )
831 if getattr(self, "_of_type", None):
832 info: Optional[_InternalEntityType[Any]] = inspect(
833 self._of_type
834 )
835 assert info is not None
836 target_mapper, to_selectable, is_aliased_class = (
837 info.mapper,
838 info.selectable,
839 info.is_aliased_class,
840 )
841 if self.property._is_self_referential and not is_aliased_class:
842 to_selectable = to_selectable._anonymous_fromclause()
844 single_crit = target_mapper._single_table_criterion
845 if single_crit is not None:
846 if where_criteria is not None:
847 where_criteria = single_crit & where_criteria
848 else:
849 where_criteria = single_crit
850 dest_entity = info
851 else:
852 is_aliased_class = False
853 to_selectable = None
854 dest_entity = self.mapper
856 if self.adapter:
857 source_selectable = self._source_selectable()
858 else:
859 source_selectable = None
861 (
862 pj,
863 sj,
864 source,
865 dest,
866 secondary,
867 target_adapter,
868 ) = self.property._create_joins(
869 dest_selectable=to_selectable,
870 source_selectable=source_selectable,
871 )
873 for k in kwargs:
874 crit = getattr(self.property.mapper.class_, k) == kwargs[k]
875 if where_criteria is None:
876 where_criteria = crit
877 else:
878 where_criteria = where_criteria & crit
880 # annotate the *local* side of the join condition, in the case
881 # of pj + sj this is the full primaryjoin, in the case of just
882 # pj its the local side of the primaryjoin.
883 j: ColumnElement[bool]
884 if sj is not None:
885 j = pj & sj
886 else:
887 j = pj
889 if (
890 where_criteria is not None
891 and target_adapter
892 and not is_aliased_class
893 ):
894 # limit this adapter to annotated only?
895 where_criteria = target_adapter.traverse(where_criteria)
897 # only have the "joined left side" of what we
898 # return be subject to Query adaption. The right
899 # side of it is used for an exists() subquery and
900 # should not correlate or otherwise reach out
901 # to anything in the enclosing query.
902 if where_criteria is not None:
903 where_criteria = where_criteria._annotate(
904 {"no_replacement_traverse": True}
905 )
907 crit = j & sql.True_._ifnone(where_criteria)
909 # ensure the exists query gets picked up by the ORM
910 # compiler and that it has what we expect as parententity so that
911 # _adjust_for_extra_criteria() gets set up
912 dest = dest._annotate(
913 {
914 "parentmapper": dest_entity.mapper,
915 "entity_namespace": dest_entity,
916 "parententity": dest_entity,
917 }
918 )._set_propagate_attrs(
919 {"compile_state_plugin": "orm", "plugin_subject": dest_entity}
920 )
921 if secondary is not None:
922 ex = (
923 sql.exists(1)
924 .where(crit)
925 .select_from(dest, secondary)
926 .correlate_except(dest, secondary)
927 )
928 else:
929 ex = (
930 sql.exists(1)
931 .where(crit)
932 .select_from(dest)
933 .correlate_except(dest)
934 )
935 return ex
937 def any(
938 self,
939 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
940 **kwargs: Any,
941 ) -> ColumnElement[bool]:
942 """Produce an expression that tests a collection against
943 particular criterion, using EXISTS.
945 An expression like::
947 session.query(MyClass).filter(
948 MyClass.somereference.any(SomeRelated.x == 2)
949 )
951 Will produce a query like:
953 .. sourcecode:: sql
955 SELECT * FROM my_table WHERE
956 EXISTS (SELECT 1 FROM related WHERE related.my_id=my_table.id
957 AND related.x=2)
959 Because :meth:`~.RelationshipProperty.Comparator.any` uses
960 a correlated subquery, its performance is not nearly as
961 good when compared against large target tables as that of
962 using a join.
964 :meth:`~.RelationshipProperty.Comparator.any` is particularly
965 useful for testing for empty collections::
967 session.query(MyClass).filter(~MyClass.somereference.any())
969 will produce:
971 .. sourcecode:: sql
973 SELECT * FROM my_table WHERE
974 NOT (EXISTS (SELECT 1 FROM related WHERE
975 related.my_id=my_table.id))
977 :meth:`~.RelationshipProperty.Comparator.any` is only
978 valid for collections, i.e. a :func:`_orm.relationship`
979 that has ``uselist=True``. For scalar references,
980 use :meth:`~.RelationshipProperty.Comparator.has`.
982 """
983 if not self.property.uselist:
984 raise sa_exc.InvalidRequestError(
985 "'any()' not implemented for scalar "
986 "attributes. Use has()."
987 )
989 return self._criterion_exists(criterion, **kwargs)
991 def has(
992 self,
993 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
994 **kwargs: Any,
995 ) -> ColumnElement[bool]:
996 """Produce an expression that tests a scalar reference against
997 particular criterion, using EXISTS.
999 An expression like::
1001 session.query(MyClass).filter(
1002 MyClass.somereference.has(SomeRelated.x == 2)
1003 )
1005 Will produce a query like:
1007 .. sourcecode:: sql
1009 SELECT * FROM my_table WHERE
1010 EXISTS (SELECT 1 FROM related WHERE
1011 related.id==my_table.related_id AND related.x=2)
1013 Because :meth:`~.RelationshipProperty.Comparator.has` uses
1014 a correlated subquery, its performance is not nearly as
1015 good when compared against large target tables as that of
1016 using a join.
1018 :meth:`~.RelationshipProperty.Comparator.has` is only
1019 valid for scalar references, i.e. a :func:`_orm.relationship`
1020 that has ``uselist=False``. For collection references,
1021 use :meth:`~.RelationshipProperty.Comparator.any`.
1023 """
1024 if self.property.uselist:
1025 raise sa_exc.InvalidRequestError(
1026 "'has()' not implemented for collections. Use any()."
1027 )
1028 return self._criterion_exists(criterion, **kwargs)
1030 def contains(
1031 self, other: _ColumnExpressionArgument[Any], **kwargs: Any
1032 ) -> ColumnElement[bool]:
1033 """Return a simple expression that tests a collection for
1034 containment of a particular item.
1036 :meth:`~.RelationshipProperty.Comparator.contains` is
1037 only valid for a collection, i.e. a
1038 :func:`_orm.relationship` that implements
1039 one-to-many or many-to-many with ``uselist=True``.
1041 When used in a simple one-to-many context, an
1042 expression like::
1044 MyClass.contains(other)
1046 Produces a clause like:
1048 .. sourcecode:: sql
1050 mytable.id == <some id>
1052 Where ``<some id>`` is the value of the foreign key
1053 attribute on ``other`` which refers to the primary
1054 key of its parent object. From this it follows that
1055 :meth:`~.RelationshipProperty.Comparator.contains` is
1056 very useful when used with simple one-to-many
1057 operations.
1059 For many-to-many operations, the behavior of
1060 :meth:`~.RelationshipProperty.Comparator.contains`
1061 has more caveats. The association table will be
1062 rendered in the statement, producing an "implicit"
1063 join, that is, includes multiple tables in the FROM
1064 clause which are equated in the WHERE clause::
1066 query(MyClass).filter(MyClass.contains(other))
1068 Produces a query like:
1070 .. sourcecode:: sql
1072 SELECT * FROM my_table, my_association_table AS
1073 my_association_table_1 WHERE
1074 my_table.id = my_association_table_1.parent_id
1075 AND my_association_table_1.child_id = <some id>
1077 Where ``<some id>`` would be the primary key of
1078 ``other``. From the above, it is clear that
1079 :meth:`~.RelationshipProperty.Comparator.contains`
1080 will **not** work with many-to-many collections when
1081 used in queries that move beyond simple AND
1082 conjunctions, such as multiple
1083 :meth:`~.RelationshipProperty.Comparator.contains`
1084 expressions joined by OR. In such cases subqueries or
1085 explicit "outer joins" will need to be used instead.
1086 See :meth:`~.RelationshipProperty.Comparator.any` for
1087 a less-performant alternative using EXISTS, or refer
1088 to :meth:`_query.Query.outerjoin`
1089 as well as :ref:`orm_queryguide_joins`
1090 for more details on constructing outer joins.
1092 kwargs may be ignored by this operator but are required for API
1093 conformance.
1094 """
1095 if not self.prop.uselist:
1096 raise sa_exc.InvalidRequestError(
1097 "'contains' not implemented for scalar "
1098 "attributes. Use =="
1099 )
1101 clause = self.prop._optimized_compare(
1102 other, adapt_source=self.adapter
1103 )
1105 if self.prop.secondaryjoin is not None:
1106 clause.negation_clause = self.__negated_contains_or_equals(
1107 other
1108 )
1110 return clause
1112 def __negated_contains_or_equals(
1113 self, other: Any
1114 ) -> ColumnElement[bool]:
1115 if self.prop.direction == MANYTOONE:
1116 state = attributes.instance_state(other)
1118 def state_bindparam(
1119 local_col: ColumnElement[Any],
1120 state: InstanceState[Any],
1121 remote_col: ColumnElement[Any],
1122 ) -> BindParameter[Any]:
1123 dict_ = state.dict
1124 return sql.bindparam(
1125 local_col.key,
1126 type_=local_col.type,
1127 unique=True,
1128 callable_=self.prop._get_attr_w_warn_on_none(
1129 self.prop.mapper, state, dict_, remote_col
1130 ),
1131 )
1133 def adapt(col: _CE) -> _CE:
1134 if self.adapter:
1135 return self.adapter(col)
1136 else:
1137 return col
1139 if self.property._use_get:
1140 return sql.and_(
1141 *[
1142 sql.or_(
1143 adapt(x)
1144 != state_bindparam(adapt(x), state, y),
1145 adapt(x) == None,
1146 )
1147 for (x, y) in self.property.local_remote_pairs
1148 ]
1149 )
1151 criterion = sql.and_(
1152 *[
1153 x == y
1154 for (x, y) in zip(
1155 self.property.mapper.primary_key,
1156 self.property.mapper.primary_key_from_instance(other),
1157 )
1158 ]
1159 )
1161 return ~self._criterion_exists(criterion)
1163 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
1164 """Implement the ``!=`` operator.
1166 In a many-to-one context, such as:
1168 .. sourcecode:: text
1170 MyClass.some_prop != <some object>
1172 This will typically produce a clause such as:
1174 .. sourcecode:: sql
1176 mytable.related_id != <some id>
1178 Where ``<some id>`` is the primary key of the
1179 given object.
1181 The ``!=`` operator provides partial functionality for non-
1182 many-to-one comparisons:
1184 * Comparisons against collections are not supported.
1185 Use
1186 :meth:`~.RelationshipProperty.Comparator.contains`
1187 in conjunction with :func:`_expression.not_`.
1188 * Compared to a scalar one-to-many, will produce a
1189 clause that compares the target columns in the parent to
1190 the given target.
1191 * Compared to a scalar many-to-many, an alias
1192 of the association table will be rendered as
1193 well, forming a natural join that is part of the
1194 main body of the query. This will not work for
1195 queries that go beyond simple AND conjunctions of
1196 comparisons, such as those which use OR. Use
1197 explicit joins, outerjoins, or
1198 :meth:`~.RelationshipProperty.Comparator.has` in
1199 conjunction with :func:`_expression.not_` for
1200 more comprehensive non-many-to-one scalar
1201 membership tests.
1202 * Comparisons against ``None`` given in a one-to-many
1203 or many-to-many context produce an EXISTS clause.
1205 """
1206 if other is None or isinstance(other, expression.Null):
1207 if self.property.direction == MANYTOONE:
1208 return ~self.property._optimized_compare(
1209 None, adapt_source=self.adapter
1210 )
1212 else:
1213 return self._criterion_exists()
1214 elif self.property.uselist:
1215 raise sa_exc.InvalidRequestError(
1216 "Can't compare a collection"
1217 " to an object or collection; use "
1218 "contains() to test for membership."
1219 )
1220 else:
1221 return self.__negated_contains_or_equals(other)
1223 if TYPE_CHECKING:
1224 property: RelationshipProperty[_PT] # noqa: A001
1226 def _memoized_attr_property(self) -> RelationshipProperty[_PT]:
1227 self.prop.parent._check_configure()
1228 return self.prop
1230 def _with_parent(
1231 self,
1232 instance: object,
1233 alias_secondary: bool = True,
1234 from_entity: Optional[_EntityType[Any]] = None,
1235 ) -> ColumnElement[bool]:
1236 assert instance is not None
1237 adapt_source: Optional[_CoreAdapterProto] = None
1238 if from_entity is not None:
1239 insp: Optional[_InternalEntityType[Any]] = inspect(from_entity)
1240 assert insp is not None
1241 if insp_is_aliased_class(insp):
1242 adapt_source = insp._adapter.adapt_clause
1243 return self._optimized_compare(
1244 instance,
1245 value_is_parent=True,
1246 adapt_source=adapt_source,
1247 alias_secondary=alias_secondary,
1248 )
1250 def _optimized_compare(
1251 self,
1252 state: Any,
1253 value_is_parent: bool = False,
1254 adapt_source: Optional[_CoreAdapterProto] = None,
1255 alias_secondary: bool = True,
1256 ) -> ColumnElement[bool]:
1257 if state is not None:
1258 try:
1259 state = inspect(state)
1260 except sa_exc.NoInspectionAvailable:
1261 state = None
1263 if state is None or not getattr(state, "is_instance", False):
1264 raise sa_exc.ArgumentError(
1265 "Mapped instance expected for relationship "
1266 "comparison to object. Classes, queries and other "
1267 "SQL elements are not accepted in this context; for "
1268 "comparison with a subquery, "
1269 "use %s.has(**criteria)." % self
1270 )
1271 reverse_direction = not value_is_parent
1273 if state is None:
1274 return self._lazy_none_clause(
1275 reverse_direction, adapt_source=adapt_source
1276 )
1278 if not reverse_direction:
1279 criterion, bind_to_col = (
1280 self._lazy_strategy._lazywhere,
1281 self._lazy_strategy._bind_to_col,
1282 )
1283 else:
1284 criterion, bind_to_col = (
1285 self._lazy_strategy._rev_lazywhere,
1286 self._lazy_strategy._rev_bind_to_col,
1287 )
1289 if reverse_direction:
1290 mapper = self.mapper
1291 else:
1292 mapper = self.parent
1294 dict_ = attributes.instance_dict(state.obj())
1296 def visit_bindparam(bindparam: BindParameter[Any]) -> None:
1297 if bindparam._identifying_key in bind_to_col:
1298 bindparam.callable = self._get_attr_w_warn_on_none(
1299 mapper,
1300 state,
1301 dict_,
1302 bind_to_col[bindparam._identifying_key],
1303 )
1305 if self.secondary is not None and alias_secondary:
1306 criterion = ClauseAdapter(
1307 self.secondary._anonymous_fromclause()
1308 ).traverse(criterion)
1310 criterion = visitors.cloned_traverse(
1311 criterion, {}, {"bindparam": visit_bindparam}
1312 )
1314 if adapt_source:
1315 criterion = adapt_source(criterion)
1316 return criterion
1318 def _get_attr_w_warn_on_none(
1319 self,
1320 mapper: Mapper[Any],
1321 state: InstanceState[Any],
1322 dict_: _InstanceDict,
1323 column: ColumnElement[Any],
1324 ) -> Callable[[], Any]:
1325 """Create the callable that is used in a many-to-one expression.
1327 E.g.::
1329 u1 = s.query(User).get(5)
1331 expr = Address.user == u1
1333 Above, the SQL should be "address.user_id = 5". The callable
1334 returned by this method produces the value "5" based on the identity
1335 of ``u1``.
1337 """
1339 # in this callable, we're trying to thread the needle through
1340 # a wide variety of scenarios, including:
1341 #
1342 # * the object hasn't been flushed yet and there's no value for
1343 # the attribute as of yet
1344 #
1345 # * the object hasn't been flushed yet but it has a user-defined
1346 # value
1347 #
1348 # * the object has a value but it's expired and not locally present
1349 #
1350 # * the object has a value but it's expired and not locally present,
1351 # and the object is also detached
1352 #
1353 # * The object hadn't been flushed yet, there was no value, but
1354 # later, the object has been expired and detached, and *now*
1355 # they're trying to evaluate it
1356 #
1357 # * the object had a value, but it was changed to a new value, and
1358 # then expired
1359 #
1360 # * the object had a value, but it was changed to a new value, and
1361 # then expired, then the object was detached
1362 #
1363 # * the object has a user-set value, but it's None and we don't do
1364 # the comparison correctly for that so warn
1365 #
1367 prop = mapper.get_property_by_column(column)
1369 # by invoking this method, InstanceState will track the last known
1370 # value for this key each time the attribute is to be expired.
1371 # this feature was added explicitly for use in this method.
1372 state._track_last_known_value(prop.key)
1374 lkv_fixed = state._last_known_values
1376 def _go() -> Any:
1377 assert lkv_fixed is not None
1378 last_known = to_return = lkv_fixed[prop.key]
1379 existing_is_available = (
1380 last_known is not LoaderCallableStatus.NO_VALUE
1381 )
1383 # we support that the value may have changed. so here we
1384 # try to get the most recent value including re-fetching.
1385 # only if we can't get a value now due to detachment do we return
1386 # the last known value
1387 current_value = mapper._get_state_attr_by_column(
1388 state,
1389 dict_,
1390 column,
1391 passive=(
1392 PassiveFlag.PASSIVE_OFF
1393 if state.persistent
1394 else PassiveFlag.PASSIVE_NO_FETCH ^ PassiveFlag.INIT_OK
1395 ),
1396 )
1398 if current_value is LoaderCallableStatus.NEVER_SET:
1399 if not existing_is_available:
1400 raise sa_exc.InvalidRequestError(
1401 "Can't resolve value for column %s on object "
1402 "%s; no value has been set for this column"
1403 % (column, state_str(state))
1404 )
1405 elif current_value is LoaderCallableStatus.PASSIVE_NO_RESULT:
1406 if not existing_is_available:
1407 raise sa_exc.InvalidRequestError(
1408 "Can't resolve value for column %s on object "
1409 "%s; the object is detached and the value was "
1410 "expired" % (column, state_str(state))
1411 )
1412 else:
1413 to_return = current_value
1414 if to_return is None:
1415 util.warn(
1416 "Got None for value of column %s; this is unsupported "
1417 "for a relationship comparison and will not "
1418 "currently produce an IS comparison "
1419 "(but may in a future release)" % column
1420 )
1421 return to_return
1423 return _go
1425 def _lazy_none_clause(
1426 self,
1427 reverse_direction: bool = False,
1428 adapt_source: Optional[_CoreAdapterProto] = None,
1429 ) -> ColumnElement[bool]:
1430 if not reverse_direction:
1431 criterion, bind_to_col = (
1432 self._lazy_strategy._lazywhere,
1433 self._lazy_strategy._bind_to_col,
1434 )
1435 else:
1436 criterion, bind_to_col = (
1437 self._lazy_strategy._rev_lazywhere,
1438 self._lazy_strategy._rev_bind_to_col,
1439 )
1441 criterion = adapt_criterion_to_null(criterion, bind_to_col)
1443 if adapt_source:
1444 criterion = adapt_source(criterion)
1445 return criterion
1447 def _format_as_string(self, class_: type, key: str) -> str:
1448 return f"{class_.__name__}.{key}"
1450 def __str__(self) -> str:
1451 return self._format_as_string(self.parent.class_, self.key)
1453 def merge(
1454 self,
1455 session: Session,
1456 source_state: InstanceState[Any],
1457 source_dict: _InstanceDict,
1458 dest_state: InstanceState[Any],
1459 dest_dict: _InstanceDict,
1460 load: bool,
1461 _recursive: Dict[Any, object],
1462 _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
1463 ) -> None:
1464 if load:
1465 for r in self._reverse_property:
1466 if (source_state, r) in _recursive:
1467 return
1469 if "merge" not in self._cascade:
1470 return
1472 if self.key not in source_dict:
1473 return
1475 if self.uselist:
1476 impl = source_state.get_impl(self.key)
1478 assert is_has_collection_adapter(impl)
1479 instances_iterable = impl.get_collection(source_state, source_dict)
1481 # if this is a CollectionAttributeImpl, then empty should
1482 # be False, otherwise "self.key in source_dict" should not be
1483 # True
1484 assert not instances_iterable.empty if impl.collection else True
1486 if load:
1487 # for a full merge, pre-load the destination collection,
1488 # so that individual _merge of each item pulls from identity
1489 # map for those already present.
1490 # also assumes CollectionAttributeImpl behavior of loading
1491 # "old" list in any case
1492 dest_state.get_impl(self.key).get(
1493 dest_state, dest_dict, passive=PassiveFlag.PASSIVE_MERGE
1494 )
1496 dest_list = []
1497 for current in instances_iterable:
1498 current_state = attributes.instance_state(current)
1499 current_dict = attributes.instance_dict(current)
1500 _recursive[(current_state, self)] = True
1501 obj = session._merge(
1502 current_state,
1503 current_dict,
1504 load=load,
1505 _recursive=_recursive,
1506 _resolve_conflict_map=_resolve_conflict_map,
1507 )
1508 if obj is not None:
1509 dest_list.append(obj)
1511 if not load:
1512 coll = attributes.init_state_collection(
1513 dest_state, dest_dict, self.key
1514 )
1515 for c in dest_list:
1516 coll.append_without_event(c)
1517 else:
1518 dest_impl = dest_state.get_impl(self.key)
1519 assert is_has_collection_adapter(dest_impl)
1520 dest_impl.set(
1521 dest_state,
1522 dest_dict,
1523 dest_list,
1524 _adapt=False,
1525 passive=PassiveFlag.PASSIVE_MERGE,
1526 )
1527 else:
1528 current = source_dict[self.key]
1529 if current is not None:
1530 current_state = attributes.instance_state(current)
1531 current_dict = attributes.instance_dict(current)
1532 _recursive[(current_state, self)] = True
1533 obj = session._merge(
1534 current_state,
1535 current_dict,
1536 load=load,
1537 _recursive=_recursive,
1538 _resolve_conflict_map=_resolve_conflict_map,
1539 )
1540 else:
1541 obj = None
1543 if not load:
1544 dest_dict[self.key] = obj
1545 else:
1546 dest_state.get_impl(self.key).set(
1547 dest_state, dest_dict, obj, None
1548 )
1550 def _value_as_iterable(
1551 self,
1552 state: InstanceState[_O],
1553 dict_: _InstanceDict,
1554 key: str,
1555 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1556 ) -> Sequence[Tuple[InstanceState[_O], _O]]:
1557 """Return a list of tuples (state, obj) for the given
1558 key.
1560 returns an empty list if the value is None/empty/PASSIVE_NO_RESULT
1561 """
1563 impl = state.manager[key].impl
1564 x = impl.get(state, dict_, passive=passive)
1565 if x is LoaderCallableStatus.PASSIVE_NO_RESULT or x is None:
1566 return []
1567 elif is_has_collection_adapter(impl):
1568 return [
1569 (attributes.instance_state(o), o)
1570 for o in impl.get_collection(state, dict_, x, passive=passive)
1571 ]
1572 else:
1573 return [(attributes.instance_state(x), x)]
1575 def cascade_iterator(
1576 self,
1577 type_: str,
1578 state: InstanceState[Any],
1579 dict_: _InstanceDict,
1580 visited_states: Set[InstanceState[Any]],
1581 halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
1582 ) -> Iterator[Tuple[Any, Mapper[Any], InstanceState[Any], _InstanceDict]]:
1583 # assert type_ in self._cascade
1585 # only actively lazy load on the 'delete' cascade
1586 if type_ != "delete" or self.passive_deletes:
1587 passive = PassiveFlag.PASSIVE_NO_INITIALIZE
1588 else:
1589 passive = PassiveFlag.PASSIVE_OFF | PassiveFlag.NO_RAISE
1591 if type_ == "save-update":
1592 tuples = state.manager[self.key].impl.get_all_pending(state, dict_)
1593 else:
1594 tuples = self._value_as_iterable(
1595 state, dict_, self.key, passive=passive
1596 )
1598 skip_pending = (
1599 type_ == "refresh-expire" and "delete-orphan" not in self._cascade
1600 )
1602 for instance_state, c in tuples:
1603 if instance_state in visited_states:
1604 continue
1606 if c is None:
1607 # would like to emit a warning here, but
1608 # would not be consistent with collection.append(None)
1609 # current behavior of silently skipping.
1610 # see [ticket:2229]
1611 continue
1613 assert instance_state is not None
1614 instance_dict = attributes.instance_dict(c)
1616 if halt_on and halt_on(instance_state):
1617 continue
1619 if skip_pending and not instance_state.key:
1620 continue
1622 instance_mapper = instance_state.manager.mapper
1624 if not instance_mapper.isa(self.mapper.class_manager.mapper):
1625 raise AssertionError(
1626 "Attribute '%s' on class '%s' "
1627 "doesn't handle objects "
1628 "of type '%s'"
1629 % (self.key, self.parent.class_, c.__class__)
1630 )
1632 visited_states.add(instance_state)
1634 yield c, instance_mapper, instance_state, instance_dict
1636 @property
1637 def _effective_sync_backref(self) -> bool:
1638 if self.viewonly:
1639 return False
1640 else:
1641 return self.sync_backref is not False
1643 @staticmethod
1644 def _check_sync_backref(
1645 rel_a: RelationshipProperty[Any], rel_b: RelationshipProperty[Any]
1646 ) -> None:
1647 if rel_a.viewonly and rel_b.sync_backref:
1648 raise sa_exc.InvalidRequestError(
1649 "Relationship %s cannot specify sync_backref=True since %s "
1650 "includes viewonly=True." % (rel_b, rel_a)
1651 )
1652 if (
1653 rel_a.viewonly
1654 and not rel_b.viewonly
1655 and rel_b.sync_backref is not False
1656 ):
1657 rel_b.sync_backref = False
1659 def _add_reverse_property(self, key: str) -> None:
1660 other = self.mapper.get_property(key, _configure_mappers=False)
1661 if not isinstance(other, RelationshipProperty):
1662 raise sa_exc.InvalidRequestError(
1663 "back_populates on relationship '%s' refers to attribute '%s' "
1664 "that is not a relationship. The back_populates parameter "
1665 "should refer to the name of a relationship on the target "
1666 "class." % (self, other)
1667 )
1668 # viewonly and sync_backref cases
1669 # 1. self.viewonly==True and other.sync_backref==True -> error
1670 # 2. self.viewonly==True and other.viewonly==False and
1671 # other.sync_backref==None -> warn sync_backref=False, set to False
1672 self._check_sync_backref(self, other)
1673 # 3. other.viewonly==True and self.sync_backref==True -> error
1674 # 4. other.viewonly==True and self.viewonly==False and
1675 # self.sync_backref==None -> warn sync_backref=False, set to False
1676 self._check_sync_backref(other, self)
1678 self._reverse_property.add(other)
1679 other._reverse_property.add(self)
1681 other._setup_entity()
1683 if not other.mapper.common_parent(self.parent):
1684 raise sa_exc.ArgumentError(
1685 "reverse_property %r on "
1686 "relationship %s references relationship %s, which "
1687 "does not reference mapper %s"
1688 % (key, self, other, self.parent)
1689 )
1691 if (
1692 other._configure_started
1693 and self.direction in (ONETOMANY, MANYTOONE)
1694 and self.direction == other.direction
1695 ):
1696 raise sa_exc.ArgumentError(
1697 "%s and back-reference %s are "
1698 "both of the same direction %r. Did you mean to "
1699 "set remote_side on the many-to-one side ?"
1700 % (other, self, self.direction)
1701 )
1703 @util.memoized_property
1704 def entity(self) -> _InternalEntityType[_T]:
1705 """Return the target mapped entity, which is an inspect() of the
1706 class or aliased class that is referenced by this
1707 :class:`.RelationshipProperty`.
1709 """
1710 self.parent._check_configure()
1711 return self.entity
1713 @util.memoized_property
1714 def mapper(self) -> Mapper[_T]:
1715 """Return the targeted :class:`_orm.Mapper` for this
1716 :class:`.RelationshipProperty`.
1718 """
1719 return self.entity.mapper
1721 def do_init(self) -> None:
1722 self._process_dependent_arguments()
1723 self._setup_entity()
1724 self._setup_registry_dependencies()
1725 self._setup_join_conditions()
1726 self._check_cascade_settings(self._cascade)
1727 self._post_init()
1728 self._generate_backref()
1729 self._join_condition._warn_for_conflicting_sync_targets()
1730 super().do_init()
1731 self._lazy_strategy = cast(
1732 "_LazyLoader", self._get_strategy((("lazy", "select"),))
1733 )
1735 def _setup_registry_dependencies(self) -> None:
1736 self.parent.mapper.registry._set_depends_on(
1737 self.entity.mapper.registry
1738 )
1740 def _process_dependent_arguments(self) -> None:
1741 """Convert incoming configuration arguments to their
1742 proper form.
1744 Callables are resolved, ORM annotations removed.
1746 """
1748 # accept callables for other attributes which may require
1749 # deferred initialization. This technique is used
1750 # by declarative "string configs" and some recipes.
1751 init_args = self._init_args
1753 for attr in (
1754 "order_by",
1755 "primaryjoin",
1756 "secondaryjoin",
1757 "secondary",
1758 "foreign_keys",
1759 "remote_side",
1760 "back_populates",
1761 ):
1762 rel_arg = getattr(init_args, attr)
1764 rel_arg._resolve_against_registry(self._clsregistry_resolvers[1])
1766 # remove "annotations" which are present if mapped class
1767 # descriptors are used to create the join expression.
1768 for attr in "primaryjoin", "secondaryjoin":
1769 rel_arg = getattr(init_args, attr)
1770 val = rel_arg.resolved
1771 if val is not None:
1772 rel_arg.resolved = coercions.expect(
1773 roles.ColumnArgumentRole, val, argname=attr
1774 )
1776 secondary = init_args.secondary.resolved
1777 if secondary is not None and _is_mapped_class(secondary):
1778 raise sa_exc.ArgumentError(
1779 "secondary argument %s passed to to relationship() %s must "
1780 "be a Table object or other FROM clause; can't send a mapped "
1781 "class directly as rows in 'secondary' are persisted "
1782 "independently of a class that is mapped "
1783 "to that same table." % (secondary, self)
1784 )
1786 # ensure expressions in self.order_by, foreign_keys,
1787 # remote_side are all columns, not strings.
1788 if (
1789 init_args.order_by.resolved is not False
1790 and init_args.order_by.resolved is not None
1791 ):
1792 self.order_by = tuple(
1793 coercions.expect(
1794 roles.ColumnArgumentRole, x, argname="order_by"
1795 )
1796 for x in util.to_list(init_args.order_by.resolved)
1797 )
1798 else:
1799 self.order_by = False
1801 self._user_defined_foreign_keys = util.column_set(
1802 coercions.expect(
1803 roles.ColumnArgumentRole, x, argname="foreign_keys"
1804 )
1805 for x in util.to_column_set(init_args.foreign_keys.resolved)
1806 )
1808 self.remote_side = util.column_set(
1809 coercions.expect(
1810 roles.ColumnArgumentRole, x, argname="remote_side"
1811 )
1812 for x in util.to_column_set(init_args.remote_side.resolved)
1813 )
1815 def declarative_scan(
1816 self,
1817 decl_scan: _DeclarativeMapperConfig,
1818 registry: _RegistryType,
1819 cls: Type[Any],
1820 originating_module: Optional[str],
1821 key: str,
1822 mapped_container: Optional[Type[Mapped[Any]]],
1823 annotation: Optional[_AnnotationScanType],
1824 extracted_mapped_annotation: Optional[_AnnotationScanType],
1825 is_dataclass_field: bool,
1826 ) -> None:
1827 if extracted_mapped_annotation is None:
1828 if self.argument is None:
1829 self._raise_for_required(key, cls)
1830 else:
1831 return
1833 argument = extracted_mapped_annotation
1834 assert originating_module is not None
1836 if mapped_container is not None:
1837 is_write_only = issubclass(mapped_container, WriteOnlyMapped)
1838 is_dynamic = issubclass(mapped_container, DynamicMapped)
1839 if is_write_only:
1840 self.lazy = "write_only"
1841 self.strategy_key = (("lazy", self.lazy),)
1842 elif is_dynamic:
1843 self.lazy = "dynamic"
1844 self.strategy_key = (("lazy", self.lazy),)
1845 else:
1846 is_write_only = is_dynamic = False
1848 argument = de_optionalize_union_types(argument)
1850 if hasattr(argument, "__origin__"):
1851 arg_origin = argument.__origin__
1852 if isinstance(arg_origin, type) and issubclass(
1853 arg_origin, abc.Collection
1854 ):
1855 if self.collection_class is None:
1856 if _py_inspect.isabstract(arg_origin):
1857 raise sa_exc.ArgumentError(
1858 f"Collection annotation type {arg_origin} cannot "
1859 "be instantiated; please provide an explicit "
1860 "'collection_class' parameter "
1861 "(e.g. list, set, etc.) to the "
1862 "relationship() function to accompany this "
1863 "annotation"
1864 )
1866 self.collection_class = arg_origin
1868 elif not is_write_only and not is_dynamic:
1869 self.uselist = False
1871 if argument.__args__: # type: ignore[union-attr]
1872 if isinstance(arg_origin, type) and issubclass(
1873 arg_origin, typing.Mapping
1874 ):
1875 type_arg = argument.__args__[-1] # type: ignore[union-attr] # noqa: E501
1876 else:
1877 type_arg = argument.__args__[0] # type: ignore[union-attr]
1878 if hasattr(type_arg, "__forward_arg__"):
1879 str_argument = type_arg.__forward_arg__
1881 argument = resolve_name_to_real_class_name(
1882 str_argument, originating_module
1883 )
1884 else:
1885 argument = type_arg
1886 else:
1887 raise sa_exc.ArgumentError(
1888 f"Generic alias {argument} requires an argument"
1889 )
1890 elif hasattr(argument, "__forward_arg__"):
1891 argument = argument.__forward_arg__
1893 argument = resolve_name_to_real_class_name(
1894 argument, originating_module
1895 )
1897 if (
1898 self.collection_class is None
1899 and not is_write_only
1900 and not is_dynamic
1901 ):
1902 self.uselist = False
1904 # ticket #8759
1905 # if a lead argument was given to relationship(), like
1906 # `relationship("B")`, use that, don't replace it with class we
1907 # found in the annotation. The declarative_scan() method call here is
1908 # still useful, as we continue to derive collection type and do
1909 # checking of the annotation in any case.
1910 if self.argument is None:
1911 self.argument = cast("_RelationshipArgumentType[_T]", argument)
1913 if (
1914 self._attribute_options.dataclasses_default_factory
1915 is not _NoArg.NO_ARG
1916 ):
1917 # write only / dynamic relationships have no collection_class,
1918 # however they accept a list of objects when assigned, so
1919 # ``list`` is the expected default_factory for these
1920 expected_default_factory = (
1921 list if is_write_only or is_dynamic else self.collection_class
1922 )
1924 if (
1925 self._attribute_options.dataclasses_default_factory
1926 is not expected_default_factory
1927 ):
1928 raise sa_exc.ArgumentError(
1929 "For relationship "
1930 f"{self._format_as_string(cls, key)} using "
1931 "dataclass options, default_factory must be exactly "
1932 f"{expected_default_factory}"
1933 )
1935 @util.preload_module("sqlalchemy.orm.mapper")
1936 def _setup_entity(self, __argument: Any = None, /) -> None:
1937 if "entity" in self.__dict__:
1938 return
1940 mapperlib = util.preloaded.orm_mapper
1942 if __argument:
1943 argument = __argument
1944 else:
1945 argument = self.argument
1947 resolved_argument: _ExternalEntityType[Any]
1949 if isinstance(argument, str):
1950 # we might want to cleanup clsregistry API to make this
1951 # more straightforward
1952 resolved_argument = cast(
1953 "_ExternalEntityType[Any]",
1954 self._clsregistry_resolve_name(argument)(),
1955 )
1956 elif callable(argument) and not isinstance(
1957 argument, (type, mapperlib.Mapper)
1958 ):
1959 resolved_argument = argument()
1960 else:
1961 resolved_argument = argument
1963 entity: _InternalEntityType[Any]
1965 if isinstance(resolved_argument, type):
1966 entity = class_mapper(resolved_argument, configure=False)
1967 else:
1968 try:
1969 entity = inspect(resolved_argument)
1970 except sa_exc.NoInspectionAvailable:
1971 entity = None # type: ignore[assignment]
1973 if not hasattr(entity, "mapper"):
1974 raise sa_exc.ArgumentError(
1975 "relationship '%s' expects "
1976 "a class or a mapper argument (received: %s)"
1977 % (self.key, type(resolved_argument))
1978 )
1980 self.entity = entity
1981 self.target = self.entity.persist_selectable
1983 def _setup_join_conditions(self) -> None:
1984 self._join_condition = jc = _JoinCondition(
1985 parent_persist_selectable=self.parent.persist_selectable,
1986 child_persist_selectable=self.entity.persist_selectable,
1987 parent_local_selectable=self.parent.local_table,
1988 child_local_selectable=self.entity.local_table,
1989 primaryjoin=self._init_args.primaryjoin.resolved,
1990 secondary=self._init_args.secondary.resolved,
1991 secondaryjoin=self._init_args.secondaryjoin.resolved,
1992 parent_equivalents=self.parent._equivalent_columns,
1993 child_equivalents=self.mapper._equivalent_columns,
1994 consider_as_foreign_keys=self._user_defined_foreign_keys,
1995 local_remote_pairs=self.local_remote_pairs,
1996 remote_side=self.remote_side,
1997 self_referential=self._is_self_referential,
1998 prop=self,
1999 support_sync=not self.viewonly,
2000 can_be_synced_fn=self._columns_are_mapped,
2001 )
2002 self.primaryjoin = jc.primaryjoin
2003 self.secondaryjoin = jc.secondaryjoin
2004 self.secondary = jc.secondary
2005 self.direction = jc.direction
2006 self.local_remote_pairs = jc.local_remote_pairs
2007 self.remote_side = jc.remote_columns
2008 self.local_columns = jc.local_columns
2009 self.synchronize_pairs = jc.synchronize_pairs
2010 self._calculated_foreign_keys = jc.foreign_key_columns
2011 self.secondary_synchronize_pairs = jc.secondary_synchronize_pairs
2013 @property
2014 def _clsregistry_resolve_arg(
2015 self,
2016 ) -> Callable[[str, bool], _class_resolver]:
2017 return self._clsregistry_resolvers[1]
2019 @property
2020 def _clsregistry_resolve_name(
2021 self,
2022 ) -> Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]]:
2023 return self._clsregistry_resolvers[0]
2025 @util.memoized_property
2026 @util.preload_module("sqlalchemy.orm.clsregistry")
2027 def _clsregistry_resolvers(
2028 self,
2029 ) -> Tuple[
2030 Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]],
2031 Callable[[str, bool], _class_resolver],
2032 ]:
2033 _resolver = util.preloaded.orm_clsregistry._resolver
2035 return _resolver(self.parent.class_, self)
2037 @property
2038 def cascade(self) -> CascadeOptions:
2039 """Return the current cascade setting for this
2040 :class:`.RelationshipProperty`.
2041 """
2042 return self._cascade
2044 @cascade.setter
2045 def cascade(self, cascade: Union[str, CascadeOptions]) -> None:
2046 self._set_cascade(cascade)
2048 def _set_cascade(self, cascade_arg: Union[str, CascadeOptions]) -> None:
2049 cascade = CascadeOptions(cascade_arg)
2051 if self.viewonly:
2052 cascade = CascadeOptions(
2053 cascade.intersection(CascadeOptions._viewonly_cascades)
2054 )
2056 if "mapper" in self.__dict__:
2057 self._check_cascade_settings(cascade)
2058 self._cascade = cascade
2060 if self._dependency_processor:
2061 self._dependency_processor.cascade = cascade
2063 def _check_cascade_settings(self, cascade: CascadeOptions) -> None:
2064 if (
2065 cascade.delete_orphan
2066 and not self.single_parent
2067 and (self.direction is MANYTOMANY or self.direction is MANYTOONE)
2068 ):
2069 raise sa_exc.ArgumentError(
2070 "For %(direction)s relationship %(rel)s, delete-orphan "
2071 "cascade is normally "
2072 'configured only on the "one" side of a one-to-many '
2073 "relationship, "
2074 'and not on the "many" side of a many-to-one or many-to-many '
2075 "relationship. "
2076 "To force this relationship to allow a particular "
2077 '"%(relatedcls)s" object to be referenced by only '
2078 'a single "%(clsname)s" object at a time via the '
2079 "%(rel)s relationship, which "
2080 "would allow "
2081 "delete-orphan cascade to take place in this direction, set "
2082 "the single_parent=True flag."
2083 % {
2084 "rel": self,
2085 "direction": (
2086 "many-to-one"
2087 if self.direction is MANYTOONE
2088 else "many-to-many"
2089 ),
2090 "clsname": self.parent.class_.__name__,
2091 "relatedcls": self.mapper.class_.__name__,
2092 },
2093 code="bbf0",
2094 )
2096 if self.passive_deletes == "all" and (
2097 "delete" in cascade or "delete-orphan" in cascade
2098 ):
2099 raise sa_exc.ArgumentError(
2100 "On %s, can't set passive_deletes='all' in conjunction "
2101 "with 'delete' or 'delete-orphan' cascade" % self
2102 )
2104 if cascade.delete_orphan:
2105 self.mapper.primary_mapper()._delete_orphans.append(
2106 (self.key, self.parent.class_)
2107 )
2109 def _persists_for(self, mapper: Mapper[Any]) -> bool:
2110 """Return True if this property will persist values on behalf
2111 of the given mapper.
2113 """
2115 return (
2116 self.key in mapper.relationships
2117 and mapper.relationships[self.key] is self
2118 )
2120 def _columns_are_mapped(self, *cols: ColumnElement[Any]) -> bool:
2121 """Return True if all columns in the given collection are
2122 mapped by the tables referenced by this :class:`.RelationshipProperty`.
2124 """
2126 secondary = self._init_args.secondary.resolved
2127 for c in cols:
2128 if secondary is not None and secondary.c.contains_column(c):
2129 continue
2130 if not self.parent.persist_selectable.c.contains_column(
2131 c
2132 ) and not self.target.c.contains_column(c):
2133 return False
2134 return True
2136 def _generate_backref(self) -> None:
2137 """Interpret the 'backref' instruction to create a
2138 :func:`_orm.relationship` complementary to this one."""
2140 resolve_back_populates = self._init_args.back_populates.resolved
2142 if self.backref is not None and not resolve_back_populates:
2143 kwargs: Dict[str, Any]
2144 if isinstance(self.backref, str):
2145 backref_key, kwargs = self.backref, {}
2146 else:
2147 backref_key, kwargs = self.backref
2148 mapper = self.mapper.primary_mapper()
2150 if not mapper.concrete:
2151 check = set(mapper.iterate_to_root()).union(
2152 mapper.self_and_descendants
2153 )
2154 for m in check:
2155 if m.has_property(backref_key) and not m.concrete:
2156 raise sa_exc.ArgumentError(
2157 "Error creating backref "
2158 "'%s' on relationship '%s': property of that "
2159 "name exists on mapper '%s'"
2160 % (backref_key, self, m)
2161 )
2163 # determine primaryjoin/secondaryjoin for the
2164 # backref. Use the one we had, so that
2165 # a custom join doesn't have to be specified in
2166 # both directions.
2167 if self.secondary is not None:
2168 # for many to many, just switch primaryjoin/
2169 # secondaryjoin. use the annotated
2170 # pj/sj on the _join_condition.
2171 pj = kwargs.pop(
2172 "primaryjoin",
2173 self._join_condition.secondaryjoin_minus_local,
2174 )
2175 sj = kwargs.pop(
2176 "secondaryjoin",
2177 self._join_condition.primaryjoin_minus_local,
2178 )
2179 else:
2180 pj = kwargs.pop(
2181 "primaryjoin",
2182 self._join_condition.primaryjoin_reverse_remote,
2183 )
2184 sj = kwargs.pop("secondaryjoin", None)
2185 if sj:
2186 raise sa_exc.InvalidRequestError(
2187 "Can't assign 'secondaryjoin' on a backref "
2188 "against a non-secondary relationship."
2189 )
2191 foreign_keys = kwargs.pop(
2192 "foreign_keys", self._user_defined_foreign_keys
2193 )
2194 parent = self.parent.primary_mapper()
2195 kwargs.setdefault("viewonly", self.viewonly)
2196 kwargs.setdefault("post_update", self.post_update)
2197 kwargs.setdefault("passive_updates", self.passive_updates)
2198 kwargs.setdefault("sync_backref", self.sync_backref)
2199 self.back_populates = backref_key
2200 relationship = RelationshipProperty(
2201 parent,
2202 self.secondary,
2203 primaryjoin=pj,
2204 secondaryjoin=sj,
2205 foreign_keys=foreign_keys,
2206 back_populates=self.key,
2207 **kwargs,
2208 )
2209 mapper._configure_property(
2210 backref_key, relationship, warn_for_existing=True
2211 )
2213 if resolve_back_populates:
2214 if isinstance(resolve_back_populates, PropComparator):
2215 back_populates = resolve_back_populates.prop.key
2216 elif isinstance(resolve_back_populates, str):
2217 back_populates = resolve_back_populates
2218 else:
2219 # need test coverage for this case as well
2220 raise sa_exc.ArgumentError(
2221 f"Invalid back_populates value: {resolve_back_populates!r}"
2222 )
2224 self._add_reverse_property(back_populates)
2226 @util.preload_module("sqlalchemy.orm.dependency")
2227 def _post_init(self) -> None:
2228 dependency = util.preloaded.orm_dependency
2230 if self.uselist is None:
2231 self.uselist = self.direction is not MANYTOONE
2232 if not self.viewonly:
2233 self._dependency_processor = ( # type: ignore[no-untyped-call]
2234 dependency._DependencyProcessor.from_relationship
2235 )(self)
2237 if (
2238 self.uselist
2239 and self._attribute_options.dataclasses_default
2240 is not _NoArg.NO_ARG
2241 ):
2242 raise sa_exc.ArgumentError(
2243 f"On relationship {self}, the dataclass default for "
2244 "relationship may only be set for "
2245 "a relationship that references a scalar value, i.e. "
2246 "many-to-one or explicitly uselist=False"
2247 )
2249 @util.memoized_property
2250 def _use_get(self) -> bool:
2251 """memoize the 'use_get' attribute of this RelationshipLoader's
2252 lazyloader."""
2254 strategy = self._lazy_strategy
2255 return strategy.use_get
2257 @util.memoized_property
2258 def _is_self_referential(self) -> bool:
2259 return self.mapper.common_parent(self.parent)
2261 def _create_joins(
2262 self,
2263 source_polymorphic: bool = False,
2264 source_selectable: Optional[FromClause] = None,
2265 dest_selectable: Optional[FromClause] = None,
2266 of_type_entity: Optional[_InternalEntityType[Any]] = None,
2267 alias_secondary: bool = False,
2268 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
2269 ) -> Tuple[
2270 ColumnElement[bool],
2271 Optional[ColumnElement[bool]],
2272 FromClause,
2273 FromClause,
2274 Optional[FromClause],
2275 Optional[ClauseAdapter],
2276 ]:
2277 aliased = False
2279 if alias_secondary and self.secondary is not None:
2280 aliased = True
2282 if source_selectable is None:
2283 if source_polymorphic and self.parent.with_polymorphic:
2284 source_selectable = self.parent._with_polymorphic_selectable
2286 if of_type_entity:
2287 dest_mapper = of_type_entity.mapper
2288 if dest_selectable is None:
2289 dest_selectable = of_type_entity.selectable
2290 aliased = True
2291 else:
2292 dest_mapper = self.mapper
2294 if dest_selectable is None:
2295 dest_selectable = self.entity.selectable
2296 if self.mapper.with_polymorphic:
2297 aliased = True
2299 if self._is_self_referential and source_selectable is None:
2300 dest_selectable = dest_selectable._anonymous_fromclause()
2301 aliased = True
2302 elif (
2303 dest_selectable is not self.mapper._with_polymorphic_selectable
2304 or self.mapper.with_polymorphic
2305 ):
2306 aliased = True
2308 single_crit = dest_mapper._single_table_criterion
2309 aliased = aliased or (
2310 source_selectable is not None
2311 and (
2312 source_selectable
2313 is not self.parent._with_polymorphic_selectable
2314 or source_selectable._is_subquery
2315 )
2316 )
2318 (
2319 primaryjoin,
2320 secondaryjoin,
2321 secondary,
2322 target_adapter,
2323 dest_selectable,
2324 ) = self._join_condition.join_targets(
2325 source_selectable,
2326 dest_selectable,
2327 aliased,
2328 single_crit,
2329 extra_criteria,
2330 )
2331 if source_selectable is None:
2332 source_selectable = self.parent.local_table
2333 if dest_selectable is None:
2334 dest_selectable = self.entity.local_table
2335 return (
2336 primaryjoin,
2337 secondaryjoin,
2338 source_selectable,
2339 dest_selectable,
2340 secondary,
2341 target_adapter,
2342 )
2345def _annotate_columns(element: _CE, annotations: _AnnotationDict) -> _CE:
2346 def clone(elem: _CE) -> _CE:
2347 if isinstance(elem, expression.ColumnClause):
2348 elem = elem._annotate(annotations.copy()) # type: ignore[attr-defined] # noqa: E501
2349 elem._copy_internals(clone=clone)
2350 return elem
2352 if element is not None:
2353 element = clone(element)
2354 clone = None # type: ignore[assignment] # remove gc cycles
2355 return element
2358class _JoinCondition:
2359 primaryjoin_initial: Optional[ColumnElement[bool]]
2360 primaryjoin: ColumnElement[bool]
2361 secondaryjoin: Optional[ColumnElement[bool]]
2362 secondary: Optional[FromClause]
2363 prop: RelationshipProperty[Any]
2365 synchronize_pairs: _ColumnPairs
2366 secondary_synchronize_pairs: _ColumnPairs
2367 direction: RelationshipDirection
2369 parent_persist_selectable: FromClause
2370 child_persist_selectable: FromClause
2371 parent_local_selectable: FromClause
2372 child_local_selectable: FromClause
2374 _local_remote_pairs: Optional[_ColumnPairs]
2376 def __init__(
2377 self,
2378 parent_persist_selectable: FromClause,
2379 child_persist_selectable: FromClause,
2380 parent_local_selectable: FromClause,
2381 child_local_selectable: FromClause,
2382 *,
2383 primaryjoin: Optional[ColumnElement[bool]] = None,
2384 secondary: Optional[FromClause] = None,
2385 secondaryjoin: Optional[ColumnElement[bool]] = None,
2386 parent_equivalents: Optional[_EquivalentColumnMap] = None,
2387 child_equivalents: Optional[_EquivalentColumnMap] = None,
2388 consider_as_foreign_keys: Any = None,
2389 local_remote_pairs: Optional[_ColumnPairs] = None,
2390 remote_side: Any = None,
2391 self_referential: Any = False,
2392 prop: RelationshipProperty[Any],
2393 support_sync: bool = True,
2394 can_be_synced_fn: Callable[..., bool] = lambda *c: True,
2395 ):
2396 self.parent_persist_selectable = parent_persist_selectable
2397 self.parent_local_selectable = parent_local_selectable
2398 self.child_persist_selectable = child_persist_selectable
2399 self.child_local_selectable = child_local_selectable
2400 self.parent_equivalents = parent_equivalents
2401 self.child_equivalents = child_equivalents
2402 self.primaryjoin_initial = primaryjoin
2403 self.secondaryjoin = secondaryjoin
2404 self.secondary = secondary
2405 self.consider_as_foreign_keys = consider_as_foreign_keys
2406 self._local_remote_pairs = local_remote_pairs
2407 self._remote_side = remote_side
2408 self.prop = prop
2409 self.self_referential = self_referential
2410 self.support_sync = support_sync
2411 self.can_be_synced_fn = can_be_synced_fn
2413 self._determine_joins()
2414 assert self.primaryjoin is not None
2416 self._annotate_fks()
2417 self._annotate_remote()
2418 self._annotate_local()
2419 self._annotate_parentmapper()
2420 self._setup_pairs()
2421 self._check_foreign_cols(self.primaryjoin, True)
2422 if self.secondaryjoin is not None:
2423 self._check_foreign_cols(self.secondaryjoin, False)
2424 self._determine_direction()
2425 self._check_remote_side()
2426 self._log_joins()
2428 def _log_joins(self) -> None:
2429 log = self.prop.logger
2430 log.info("%s setup primary join %s", self.prop, self.primaryjoin)
2431 log.info("%s setup secondary join %s", self.prop, self.secondaryjoin)
2432 log.info(
2433 "%s synchronize pairs [%s]",
2434 self.prop,
2435 ",".join(
2436 "(%s => %s)" % (l, r) for (l, r) in self.synchronize_pairs
2437 ),
2438 )
2439 log.info(
2440 "%s secondary synchronize pairs [%s]",
2441 self.prop,
2442 ",".join(
2443 "(%s => %s)" % (l, r)
2444 for (l, r) in self.secondary_synchronize_pairs or []
2445 ),
2446 )
2447 log.info(
2448 "%s local/remote pairs [%s]",
2449 self.prop,
2450 ",".join(
2451 "(%s / %s)" % (l, r) for (l, r) in self.local_remote_pairs
2452 ),
2453 )
2454 log.info(
2455 "%s remote columns [%s]",
2456 self.prop,
2457 ",".join("%s" % col for col in self.remote_columns),
2458 )
2459 log.info(
2460 "%s local columns [%s]",
2461 self.prop,
2462 ",".join("%s" % col for col in self.local_columns),
2463 )
2464 log.info("%s relationship direction %s", self.prop, self.direction)
2466 def _determine_joins(self) -> None:
2467 """Determine the 'primaryjoin' and 'secondaryjoin' attributes,
2468 if not passed to the constructor already.
2470 This is based on analysis of the foreign key relationships
2471 between the parent and target mapped selectables.
2473 """
2474 if self.secondaryjoin is not None and self.secondary is None:
2475 raise sa_exc.ArgumentError(
2476 "Property %s specified with secondary "
2477 "join condition but "
2478 "no secondary argument" % self.prop
2479 )
2481 # find a join between the given mapper's mapped table and
2482 # the given table. will try the mapper's local table first
2483 # for more specificity, then if not found will try the more
2484 # general mapped table, which in the case of inheritance is
2485 # a join.
2486 try:
2487 consider_as_foreign_keys = self.consider_as_foreign_keys or None
2488 if self.secondary is not None:
2489 if self.secondaryjoin is None:
2490 self.secondaryjoin = join_condition(
2491 self.child_persist_selectable,
2492 self.secondary,
2493 a_subset=self.child_local_selectable,
2494 consider_as_foreign_keys=consider_as_foreign_keys,
2495 )
2496 if self.primaryjoin_initial is None:
2497 self.primaryjoin = join_condition(
2498 self.parent_persist_selectable,
2499 self.secondary,
2500 a_subset=self.parent_local_selectable,
2501 consider_as_foreign_keys=consider_as_foreign_keys,
2502 )
2503 else:
2504 self.primaryjoin = self.primaryjoin_initial
2505 else:
2506 if self.primaryjoin_initial is None:
2507 self.primaryjoin = join_condition(
2508 self.parent_persist_selectable,
2509 self.child_persist_selectable,
2510 a_subset=self.parent_local_selectable,
2511 consider_as_foreign_keys=consider_as_foreign_keys,
2512 )
2513 else:
2514 self.primaryjoin = self.primaryjoin_initial
2515 except sa_exc.NoForeignKeysError as nfe:
2516 if self.secondary is not None:
2517 raise sa_exc.NoForeignKeysError(
2518 "Could not determine join "
2519 "condition between parent/child tables on "
2520 "relationship %s - there are no foreign keys "
2521 "linking these tables via secondary table '%s'. "
2522 "Ensure that referencing columns are associated "
2523 "with a ForeignKey or ForeignKeyConstraint, or "
2524 "specify 'primaryjoin' and 'secondaryjoin' "
2525 "expressions." % (self.prop, self.secondary)
2526 ) from nfe
2527 else:
2528 raise sa_exc.NoForeignKeysError(
2529 "Could not determine join "
2530 "condition between parent/child tables on "
2531 "relationship %s - there are no foreign keys "
2532 "linking these tables. "
2533 "Ensure that referencing columns are associated "
2534 "with a ForeignKey or ForeignKeyConstraint, or "
2535 "specify a 'primaryjoin' expression." % self.prop
2536 ) from nfe
2537 except sa_exc.AmbiguousForeignKeysError as afe:
2538 if self.secondary is not None:
2539 raise sa_exc.AmbiguousForeignKeysError(
2540 "Could not determine join "
2541 "condition between parent/child tables on "
2542 "relationship %s - there are multiple foreign key "
2543 "paths linking the tables via secondary table '%s'. "
2544 "Specify the 'foreign_keys' "
2545 "argument, providing a list of those columns which "
2546 "should be counted as containing a foreign key "
2547 "reference from the secondary table to each of the "
2548 "parent and child tables." % (self.prop, self.secondary)
2549 ) from afe
2550 else:
2551 raise sa_exc.AmbiguousForeignKeysError(
2552 "Could not determine join "
2553 "condition between parent/child tables on "
2554 "relationship %s - there are multiple foreign key "
2555 "paths linking the tables. Specify the "
2556 "'foreign_keys' argument, providing a list of those "
2557 "columns which should be counted as containing a "
2558 "foreign key reference to the parent table." % self.prop
2559 ) from afe
2561 @property
2562 def primaryjoin_minus_local(self) -> ColumnElement[bool]:
2563 return _deep_deannotate(self.primaryjoin, values=("local", "remote"))
2565 @property
2566 def secondaryjoin_minus_local(self) -> ColumnElement[bool]:
2567 assert self.secondaryjoin is not None
2568 return _deep_deannotate(self.secondaryjoin, values=("local", "remote"))
2570 @util.memoized_property
2571 def primaryjoin_reverse_remote(self) -> ColumnElement[bool]:
2572 """Return the primaryjoin condition suitable for the
2573 "reverse" direction.
2575 If the primaryjoin was delivered here with pre-existing
2576 "remote" annotations, the local/remote annotations
2577 are reversed. Otherwise, the local/remote annotations
2578 are removed.
2580 """
2581 if self._has_remote_annotations:
2583 def replace(element: _CE, **kw: Any) -> Optional[_CE]:
2584 if "remote" in element._annotations:
2585 v = dict(element._annotations)
2586 del v["remote"]
2587 v["local"] = True
2588 return element._with_annotations(v)
2589 elif "local" in element._annotations:
2590 v = dict(element._annotations)
2591 del v["local"]
2592 v["remote"] = True
2593 return element._with_annotations(v)
2595 return None
2597 return visitors.replacement_traverse(self.primaryjoin, {}, replace)
2598 else:
2599 if self._has_foreign_annotations:
2600 # TODO: coverage
2601 return _deep_deannotate(
2602 self.primaryjoin, values=("local", "remote")
2603 )
2604 else:
2605 return _deep_deannotate(self.primaryjoin)
2607 def _has_annotation(self, clause: ClauseElement, annotation: str) -> bool:
2608 for col in visitors.iterate(clause, {}):
2609 if annotation in col._annotations:
2610 return True
2611 else:
2612 return False
2614 @util.memoized_property
2615 def _has_foreign_annotations(self) -> bool:
2616 return self._has_annotation(self.primaryjoin, "foreign")
2618 @util.memoized_property
2619 def _has_remote_annotations(self) -> bool:
2620 return self._has_annotation(self.primaryjoin, "remote")
2622 @util.memoized_property
2623 def secondary_covers_parent_primary_key(self) -> bool:
2624 """Return True if the "secondary" selectable's join to the parent
2625 table contains columns that encompass the complete primary key
2626 value of the parent.
2628 Used in optimizing the selectinload loader strategy to indicate
2629 the parent table need not be included in the query, as a complete
2630 primary key can be derived from the secondary table.
2632 """
2633 if self.secondary is None:
2634 return False
2636 secondary_synced_parent_cols = util.column_set(
2637 l for (l, _) in self.synchronize_pairs
2638 )
2639 return self.prop.parent._local_pk_cols.issubset(
2640 secondary_synced_parent_cols
2641 )
2643 def _annotate_fks(self) -> None:
2644 """Annotate the primaryjoin and secondaryjoin
2645 structures with 'foreign' annotations marking columns
2646 considered as foreign.
2648 """
2649 if self._has_foreign_annotations:
2650 return
2652 if self.consider_as_foreign_keys:
2653 self._annotate_from_fk_list()
2654 else:
2655 self._annotate_present_fks()
2657 def _annotate_from_fk_list(self) -> None:
2658 def check_fk(element: _CE, **kw: Any) -> Optional[_CE]:
2659 if element in self.consider_as_foreign_keys:
2660 return element._annotate({"foreign": True})
2661 return None
2663 self.primaryjoin = visitors.replacement_traverse(
2664 self.primaryjoin, {}, check_fk
2665 )
2666 if self.secondaryjoin is not None:
2667 self.secondaryjoin = visitors.replacement_traverse(
2668 self.secondaryjoin, {}, check_fk
2669 )
2671 def _annotate_present_fks(self) -> None:
2672 if self.secondary is not None:
2673 secondarycols = util.column_set(self.secondary.c)
2674 else:
2675 secondarycols = set()
2677 def is_foreign(
2678 a: ColumnElement[Any], b: ColumnElement[Any]
2679 ) -> Optional[ColumnElement[Any]]:
2680 if isinstance(a, schema.Column) and isinstance(b, schema.Column):
2681 if a.references(b):
2682 return a
2683 elif b.references(a):
2684 return b
2686 if secondarycols:
2687 if a in secondarycols and b not in secondarycols:
2688 return a
2689 elif b in secondarycols and a not in secondarycols:
2690 return b
2692 return None
2694 def visit_binary(binary: BinaryExpression[Any]) -> None:
2695 if not isinstance(
2696 binary.left, sql.ColumnElement
2697 ) or not isinstance(binary.right, sql.ColumnElement):
2698 return
2700 if (
2701 "foreign" not in binary.left._annotations
2702 and "foreign" not in binary.right._annotations
2703 ):
2704 col = is_foreign(binary.left, binary.right)
2705 if col is not None:
2706 if col.compare(binary.left):
2707 binary.left = binary.left._annotate({"foreign": True})
2708 elif col.compare(binary.right):
2709 binary.right = binary.right._annotate(
2710 {"foreign": True}
2711 )
2713 self.primaryjoin = visitors.cloned_traverse(
2714 self.primaryjoin, {}, {"binary": visit_binary}
2715 )
2716 if self.secondaryjoin is not None:
2717 self.secondaryjoin = visitors.cloned_traverse(
2718 self.secondaryjoin, {}, {"binary": visit_binary}
2719 )
2721 def _refers_to_parent_table(self) -> bool:
2722 """Return True if the join condition contains column
2723 comparisons where both columns are in both tables.
2725 """
2726 pt = self.parent_persist_selectable
2727 mt = self.child_persist_selectable
2728 result = False
2730 def visit_binary(binary: BinaryExpression[Any]) -> None:
2731 nonlocal result
2732 c, f = binary.left, binary.right
2733 if (
2734 isinstance(c, expression.ColumnClause)
2735 and isinstance(f, expression.ColumnClause)
2736 and pt.is_derived_from(c.table)
2737 and pt.is_derived_from(f.table)
2738 and mt.is_derived_from(c.table)
2739 and mt.is_derived_from(f.table)
2740 ):
2741 result = True
2743 visitors.traverse(self.primaryjoin, {}, {"binary": visit_binary})
2744 return result
2746 def _tables_overlap(self) -> bool:
2747 """Return True if parent/child tables have some overlap."""
2749 return selectables_overlap(
2750 self.parent_persist_selectable, self.child_persist_selectable
2751 )
2753 def _annotate_remote(self) -> None:
2754 """Annotate the primaryjoin and secondaryjoin
2755 structures with 'remote' annotations marking columns
2756 considered as part of the 'remote' side.
2758 """
2759 if self._has_remote_annotations:
2760 return
2762 if self.secondary is not None:
2763 self._annotate_remote_secondary()
2764 elif self._local_remote_pairs or self._remote_side:
2765 self._annotate_remote_from_args()
2766 elif self._refers_to_parent_table():
2767 self._annotate_selfref(
2768 lambda col: "foreign" in col._annotations, False
2769 )
2770 elif self._tables_overlap():
2771 self._annotate_remote_with_overlap()
2772 else:
2773 self._annotate_remote_distinct_selectables()
2775 def _annotate_remote_secondary(self) -> None:
2776 """annotate 'remote' in primaryjoin, secondaryjoin
2777 when 'secondary' is present.
2779 """
2781 assert self.secondary is not None
2782 fixed_secondary = self.secondary
2784 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2785 if fixed_secondary.c.contains_column(element):
2786 return element._annotate({"remote": True})
2787 return None
2789 self.primaryjoin = visitors.replacement_traverse(
2790 self.primaryjoin, {}, repl
2791 )
2793 assert self.secondaryjoin is not None
2794 self.secondaryjoin = visitors.replacement_traverse(
2795 self.secondaryjoin, {}, repl
2796 )
2798 def _annotate_selfref(
2799 self, fn: Callable[[ColumnElement[Any]], bool], remote_side_given: bool
2800 ) -> None:
2801 """annotate 'remote' in primaryjoin, secondaryjoin
2802 when the relationship is detected as self-referential.
2804 """
2806 def visit_binary(binary: BinaryExpression[Any]) -> None:
2807 equated = binary.left.compare(binary.right)
2808 if isinstance(binary.left, expression.ColumnClause) and isinstance(
2809 binary.right, expression.ColumnClause
2810 ):
2811 # assume one to many - FKs are "remote"
2812 if fn(binary.left):
2813 binary.left = binary.left._annotate({"remote": True})
2814 if fn(binary.right) and not equated:
2815 binary.right = binary.right._annotate({"remote": True})
2816 elif not remote_side_given:
2817 self._warn_non_column_elements()
2819 self.primaryjoin = visitors.cloned_traverse(
2820 self.primaryjoin, {}, {"binary": visit_binary}
2821 )
2823 def _annotate_remote_from_args(self) -> None:
2824 """annotate 'remote' in primaryjoin, secondaryjoin
2825 when the 'remote_side' or '_local_remote_pairs'
2826 arguments are used.
2828 """
2829 if self._local_remote_pairs:
2830 if self._remote_side:
2831 raise sa_exc.ArgumentError(
2832 "remote_side argument is redundant "
2833 "against more detailed _local_remote_side "
2834 "argument."
2835 )
2837 remote_side = [r for (l, r) in self._local_remote_pairs]
2838 else:
2839 remote_side = self._remote_side
2841 if self._refers_to_parent_table():
2842 self._annotate_selfref(lambda col: col in remote_side, True)
2843 else:
2845 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2846 # use set() to avoid generating ``__eq__()`` expressions
2847 # against each element
2848 if element in set(remote_side):
2849 return element._annotate({"remote": True})
2850 return None
2852 self.primaryjoin = visitors.replacement_traverse(
2853 self.primaryjoin, {}, repl
2854 )
2856 def _annotate_remote_with_overlap(self) -> None:
2857 """annotate 'remote' in primaryjoin, secondaryjoin
2858 when the parent/child tables have some set of
2859 tables in common, though is not a fully self-referential
2860 relationship.
2862 """
2864 def visit_binary(binary: BinaryExpression[Any]) -> None:
2865 binary.left, binary.right = proc_left_right(
2866 binary.left, binary.right
2867 )
2868 binary.right, binary.left = proc_left_right(
2869 binary.right, binary.left
2870 )
2872 check_entities = (
2873 self.prop is not None and self.prop.mapper is not self.prop.parent
2874 )
2876 def proc_left_right(
2877 left: ColumnElement[Any], right: ColumnElement[Any]
2878 ) -> Tuple[ColumnElement[Any], ColumnElement[Any]]:
2879 if isinstance(left, expression.ColumnClause) and isinstance(
2880 right, expression.ColumnClause
2881 ):
2882 if self.child_persist_selectable.c.contains_column(
2883 right
2884 ) and self.parent_persist_selectable.c.contains_column(left):
2885 right = right._annotate({"remote": True})
2886 elif (
2887 check_entities
2888 and right._annotations.get("parentmapper") is self.prop.mapper
2889 ):
2890 right = right._annotate({"remote": True})
2891 elif (
2892 check_entities
2893 and left._annotations.get("parentmapper") is self.prop.mapper
2894 ):
2895 left = left._annotate({"remote": True})
2896 else:
2897 self._warn_non_column_elements()
2899 return left, right
2901 self.primaryjoin = visitors.cloned_traverse(
2902 self.primaryjoin, {}, {"binary": visit_binary}
2903 )
2905 def _annotate_remote_distinct_selectables(self) -> None:
2906 """annotate 'remote' in primaryjoin, secondaryjoin
2907 when the parent/child tables are entirely
2908 separate.
2910 """
2912 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2913 if self.child_persist_selectable.c.contains_column(element) and (
2914 not self.parent_local_selectable.c.contains_column(element)
2915 or self.child_local_selectable.c.contains_column(element)
2916 ):
2917 return element._annotate({"remote": True})
2918 return None
2920 self.primaryjoin = visitors.replacement_traverse(
2921 self.primaryjoin, {}, repl
2922 )
2924 def _warn_non_column_elements(self) -> None:
2925 util.warn(
2926 "Non-simple column elements in primary "
2927 "join condition for property %s - consider using "
2928 "remote() annotations to mark the remote side." % self.prop
2929 )
2931 def _annotate_local(self) -> None:
2932 """Annotate the primaryjoin and secondaryjoin
2933 structures with 'local' annotations.
2935 This annotates all column elements found
2936 simultaneously in the parent table
2937 and the join condition that don't have a
2938 'remote' annotation set up from
2939 _annotate_remote() or user-defined.
2941 """
2942 if self._has_annotation(self.primaryjoin, "local"):
2943 return
2945 if self._local_remote_pairs:
2946 local_side = util.column_set(
2947 [l for (l, r) in self._local_remote_pairs]
2948 )
2949 else:
2950 local_side = util.column_set(self.parent_persist_selectable.c)
2952 def locals_(element: _CE, **kw: Any) -> Optional[_CE]:
2953 if "remote" not in element._annotations and element in local_side:
2954 return element._annotate({"local": True})
2955 return None
2957 self.primaryjoin = visitors.replacement_traverse(
2958 self.primaryjoin, {}, locals_
2959 )
2961 def _annotate_parentmapper(self) -> None:
2962 def parentmappers_(element: _CE, **kw: Any) -> Optional[_CE]:
2963 if "remote" in element._annotations:
2964 return element._annotate({"parentmapper": self.prop.mapper})
2965 elif "local" in element._annotations:
2966 return element._annotate({"parentmapper": self.prop.parent})
2967 return None
2969 self.primaryjoin = visitors.replacement_traverse(
2970 self.primaryjoin, {}, parentmappers_
2971 )
2973 def _check_remote_side(self) -> None:
2974 if not self.local_remote_pairs:
2975 raise sa_exc.ArgumentError(
2976 "Relationship %s could "
2977 "not determine any unambiguous local/remote column "
2978 "pairs based on join condition and remote_side "
2979 "arguments. "
2980 "Consider using the remote() annotation to "
2981 "accurately mark those elements of the join "
2982 "condition that are on the remote side of "
2983 "the relationship." % (self.prop,)
2984 )
2985 else:
2986 not_target = util.column_set(
2987 self.parent_persist_selectable.c
2988 ).difference(self.child_persist_selectable.c)
2990 for _, rmt in self.local_remote_pairs:
2991 if rmt in not_target:
2992 util.warn(
2993 "Expression %s is marked as 'remote', but these "
2994 "column(s) are local to the local side. The "
2995 "remote() annotation is needed only for a "
2996 "self-referential relationship where both sides "
2997 "of the relationship refer to the same tables."
2998 % (rmt,)
2999 )
3001 def _check_foreign_cols(
3002 self, join_condition: ColumnElement[bool], primary: bool
3003 ) -> None:
3004 """Check the foreign key columns collected and emit error
3005 messages."""
3006 foreign_cols = self._gather_columns_with_annotation(
3007 join_condition, "foreign"
3008 )
3010 has_foreign = bool(foreign_cols)
3012 if primary:
3013 can_sync = bool(self.synchronize_pairs)
3014 else:
3015 can_sync = bool(self.secondary_synchronize_pairs)
3017 if (
3018 self.support_sync
3019 and can_sync
3020 or (not self.support_sync and has_foreign)
3021 ):
3022 return
3024 # from here below is just determining the best error message
3025 # to report. Check for a join condition using any operator
3026 # (not just ==), perhaps they need to turn on "viewonly=True".
3027 if self.support_sync and has_foreign and not can_sync:
3028 err = (
3029 "Could not locate any simple equality expressions "
3030 "involving locally mapped foreign key columns for "
3031 "%s join condition "
3032 "'%s' on relationship %s."
3033 % (
3034 primary and "primary" or "secondary",
3035 join_condition,
3036 self.prop,
3037 )
3038 )
3039 err += (
3040 " Ensure that referencing columns are associated "
3041 "with a ForeignKey or ForeignKeyConstraint, or are "
3042 "annotated in the join condition with the foreign() "
3043 "annotation. To allow comparison operators other than "
3044 "'==', the relationship can be marked as viewonly=True."
3045 )
3047 raise sa_exc.ArgumentError(err)
3048 else:
3049 err = (
3050 "Could not locate any relevant foreign key columns "
3051 "for %s join condition '%s' on relationship %s."
3052 % (
3053 primary and "primary" or "secondary",
3054 join_condition,
3055 self.prop,
3056 )
3057 )
3058 err += (
3059 " Ensure that referencing columns are associated "
3060 "with a ForeignKey or ForeignKeyConstraint, or are "
3061 "annotated in the join condition with the foreign() "
3062 "annotation."
3063 )
3064 raise sa_exc.ArgumentError(err)
3066 def _determine_direction(self) -> None:
3067 """Determine if this relationship is one to many, many to one,
3068 many to many.
3070 """
3071 if self.secondaryjoin is not None:
3072 self.direction = MANYTOMANY
3073 else:
3074 parentcols = util.column_set(self.parent_persist_selectable.c)
3075 targetcols = util.column_set(self.child_persist_selectable.c)
3077 # fk collection which suggests ONETOMANY.
3078 onetomany_fk = targetcols.intersection(self.foreign_key_columns)
3080 # fk collection which suggests MANYTOONE.
3082 manytoone_fk = parentcols.intersection(self.foreign_key_columns)
3084 if onetomany_fk and manytoone_fk:
3085 # fks on both sides. test for overlap of local/remote
3086 # with foreign key.
3087 # we will gather columns directly from their annotations
3088 # without deannotating, so that we can distinguish on a column
3089 # that refers to itself.
3091 # 1. columns that are both remote and FK suggest
3092 # onetomany.
3093 onetomany_local = self._gather_columns_with_annotation(
3094 self.primaryjoin, "remote", "foreign"
3095 )
3097 # 2. columns that are FK but are not remote (e.g. local)
3098 # suggest manytoone.
3099 manytoone_local = {
3100 c
3101 for c in self._gather_columns_with_annotation(
3102 self.primaryjoin, "foreign"
3103 )
3104 if "remote" not in c._annotations
3105 }
3107 # 3. if both collections are present, remove columns that
3108 # refer to themselves. This is for the case of
3109 # and_(Me.id == Me.remote_id, Me.version == Me.version)
3110 if onetomany_local and manytoone_local:
3111 self_equated = self.remote_columns.intersection(
3112 self.local_columns
3113 )
3114 onetomany_local = onetomany_local.difference(self_equated)
3115 manytoone_local = manytoone_local.difference(self_equated)
3117 # at this point, if only one or the other collection is
3118 # present, we know the direction, otherwise it's still
3119 # ambiguous.
3121 if onetomany_local and not manytoone_local:
3122 self.direction = ONETOMANY
3123 elif manytoone_local and not onetomany_local:
3124 self.direction = MANYTOONE
3125 else:
3126 raise sa_exc.ArgumentError(
3127 "Can't determine relationship"
3128 " direction for relationship '%s' - foreign "
3129 "key columns within the join condition are present "
3130 "in both the parent and the child's mapped tables. "
3131 "Ensure that only those columns referring "
3132 "to a parent column are marked as foreign, "
3133 "either via the foreign() annotation or "
3134 "via the foreign_keys argument." % self.prop
3135 )
3136 elif onetomany_fk:
3137 self.direction = ONETOMANY
3138 elif manytoone_fk:
3139 self.direction = MANYTOONE
3140 else:
3141 raise sa_exc.ArgumentError(
3142 "Can't determine relationship "
3143 "direction for relationship '%s' - foreign "
3144 "key columns are present in neither the parent "
3145 "nor the child's mapped tables" % self.prop
3146 )
3148 def _deannotate_pairs(
3149 self, collection: _ColumnPairIterable
3150 ) -> _MutableColumnPairs:
3151 """provide deannotation for the various lists of
3152 pairs, so that using them in hashes doesn't incur
3153 high-overhead __eq__() comparisons against
3154 original columns mapped.
3156 """
3157 return [(x._deannotate(), y._deannotate()) for x, y in collection]
3159 def _setup_pairs(self) -> None:
3160 sync_pairs: _MutableColumnPairs = []
3161 lrp: util.OrderedSet[Tuple[ColumnElement[Any], ColumnElement[Any]]] = (
3162 util.OrderedSet([])
3163 )
3164 secondary_sync_pairs: _MutableColumnPairs = []
3166 def go(
3167 joincond: ColumnElement[bool],
3168 collection: _MutableColumnPairs,
3169 ) -> None:
3170 def visit_binary(
3171 binary: BinaryExpression[Any],
3172 left: ColumnElement[Any],
3173 right: ColumnElement[Any],
3174 ) -> None:
3175 if (
3176 "remote" in right._annotations
3177 and "remote" not in left._annotations
3178 and self.can_be_synced_fn(left)
3179 ):
3180 lrp.add((left, right))
3181 elif (
3182 "remote" in left._annotations
3183 and "remote" not in right._annotations
3184 and self.can_be_synced_fn(right)
3185 ):
3186 lrp.add((right, left))
3187 if binary.operator is operators.eq and self.can_be_synced_fn(
3188 left, right
3189 ):
3190 if "foreign" in right._annotations:
3191 collection.append((left, right))
3192 elif "foreign" in left._annotations:
3193 collection.append((right, left))
3195 visit_binary_product(visit_binary, joincond)
3197 for joincond, collection in [
3198 (self.primaryjoin, sync_pairs),
3199 (self.secondaryjoin, secondary_sync_pairs),
3200 ]:
3201 if joincond is None:
3202 continue
3203 go(joincond, collection)
3205 self.local_remote_pairs = self._deannotate_pairs(lrp)
3206 self.synchronize_pairs = self._deannotate_pairs(sync_pairs)
3207 self.secondary_synchronize_pairs = self._deannotate_pairs(
3208 secondary_sync_pairs
3209 )
3211 _track_overlapping_sync_targets: weakref.WeakKeyDictionary[
3212 ColumnElement[Any],
3213 weakref.WeakKeyDictionary[
3214 RelationshipProperty[Any], ColumnElement[Any]
3215 ],
3216 ] = weakref.WeakKeyDictionary()
3218 def _warn_for_conflicting_sync_targets(self) -> None:
3219 if not self.support_sync:
3220 return
3222 # we would like to detect if we are synchronizing any column
3223 # pairs in conflict with another relationship that wishes to sync
3224 # an entirely different column to the same target. This is a
3225 # very rare edge case so we will try to minimize the memory/overhead
3226 # impact of this check
3227 for from_, to_ in [
3228 (from_, to_) for (from_, to_) in self.synchronize_pairs
3229 ] + [
3230 (from_, to_) for (from_, to_) in self.secondary_synchronize_pairs
3231 ]:
3232 # save ourselves a ton of memory and overhead by only
3233 # considering columns that are subject to a overlapping
3234 # FK constraints at the core level. This condition can arise
3235 # if multiple relationships overlap foreign() directly, but
3236 # we're going to assume it's typically a ForeignKeyConstraint-
3237 # level configuration that benefits from this warning.
3239 if to_ not in self._track_overlapping_sync_targets:
3240 self._track_overlapping_sync_targets[to_] = (
3241 weakref.WeakKeyDictionary({self.prop: from_})
3242 )
3243 else:
3244 other_props = []
3245 prop_to_from = self._track_overlapping_sync_targets[to_]
3247 for pr, fr_ in prop_to_from.items():
3248 if (
3249 not pr.mapper._dispose_called
3250 and pr not in self.prop._reverse_property
3251 and pr.key not in self.prop._overlaps
3252 and self.prop.key not in pr._overlaps
3253 # note: the "__*" symbol is used internally by
3254 # SQLAlchemy as a general means of suppressing the
3255 # overlaps warning for some extension cases, however
3256 # this is not currently
3257 # a publicly supported symbol and may change at
3258 # any time.
3259 and "__*" not in self.prop._overlaps
3260 and "__*" not in pr._overlaps
3261 and not self.prop.parent.is_sibling(pr.parent)
3262 and not self.prop.mapper.is_sibling(pr.mapper)
3263 and not self.prop.parent.is_sibling(pr.mapper)
3264 and not self.prop.mapper.is_sibling(pr.parent)
3265 and (
3266 self.prop.key != pr.key
3267 or not self.prop.parent.common_parent(pr.parent)
3268 )
3269 ):
3270 other_props.append((pr, fr_))
3272 if other_props:
3273 util.warn(
3274 "relationship '%s' will copy column %s to column %s, "
3275 "which conflicts with relationship(s): %s. "
3276 "If this is not the intention, consider if these "
3277 "relationships should be linked with "
3278 "back_populates, or if viewonly=True should be "
3279 "applied to one or more if they are read-only. "
3280 "For the less common case that foreign key "
3281 "constraints are partially overlapping, the "
3282 "orm.foreign() "
3283 "annotation can be used to isolate the columns that "
3284 "should be written towards. To silence this "
3285 "warning, add the parameter 'overlaps=\"%s\"' to the "
3286 "'%s' relationship."
3287 % (
3288 self.prop,
3289 from_,
3290 to_,
3291 ", ".join(
3292 sorted(
3293 "'%s' (copies %s to %s)" % (pr, fr_, to_)
3294 for (pr, fr_) in other_props
3295 )
3296 ),
3297 ",".join(sorted(pr.key for pr, fr in other_props)),
3298 self.prop,
3299 ),
3300 code="qzyx",
3301 )
3302 self._track_overlapping_sync_targets[to_][self.prop] = from_
3304 @util.memoized_property
3305 def remote_columns(self) -> Set[ColumnElement[Any]]:
3306 return self._gather_join_annotations("remote")
3308 @util.memoized_property
3309 def local_columns(self) -> Set[ColumnElement[Any]]:
3310 return self._gather_join_annotations("local")
3312 @util.memoized_property
3313 def foreign_key_columns(self) -> Set[ColumnElement[Any]]:
3314 return self._gather_join_annotations("foreign")
3316 def _gather_join_annotations(
3317 self, annotation: str
3318 ) -> Set[ColumnElement[Any]]:
3319 s = set(
3320 self._gather_columns_with_annotation(self.primaryjoin, annotation)
3321 )
3322 if self.secondaryjoin is not None:
3323 s.update(
3324 self._gather_columns_with_annotation(
3325 self.secondaryjoin, annotation
3326 )
3327 )
3328 return {x._deannotate() for x in s}
3330 def _gather_columns_with_annotation(
3331 self, clause: ColumnElement[Any], *annotation: Iterable[str]
3332 ) -> Set[ColumnElement[Any]]:
3333 annotation_set = set(annotation)
3334 return {
3335 cast(ColumnElement[Any], col)
3336 for col in visitors.iterate(clause, {})
3337 if annotation_set.issubset(col._annotations)
3338 }
3340 @util.memoized_property
3341 def _secondary_lineage_set(self) -> FrozenSet[ColumnElement[Any]]:
3342 if self.secondary is not None:
3343 return frozenset(
3344 itertools.chain(*[c.proxy_set for c in self.secondary.c])
3345 )
3346 else:
3347 return util.EMPTY_SET
3349 def join_targets(
3350 self,
3351 source_selectable: Optional[FromClause],
3352 dest_selectable: FromClause,
3353 aliased: bool,
3354 single_crit: Optional[ColumnElement[bool]] = None,
3355 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
3356 ) -> Tuple[
3357 ColumnElement[bool],
3358 Optional[ColumnElement[bool]],
3359 Optional[FromClause],
3360 Optional[ClauseAdapter],
3361 FromClause,
3362 ]:
3363 """Given a source and destination selectable, create a
3364 join between them.
3366 This takes into account aliasing the join clause
3367 to reference the appropriate corresponding columns
3368 in the target objects, as well as the extra child
3369 criterion, equivalent column sets, etc.
3371 """
3372 # place a barrier on the destination such that
3373 # replacement traversals won't ever dig into it.
3374 # its internal structure remains fixed
3375 # regardless of context.
3376 dest_selectable = _shallow_annotate(
3377 dest_selectable, {"no_replacement_traverse": True}
3378 )
3380 primaryjoin, secondaryjoin, secondary = (
3381 self.primaryjoin,
3382 self.secondaryjoin,
3383 self.secondary,
3384 )
3386 # adjust the join condition for single table inheritance,
3387 # in the case that the join is to a subclass
3388 # this is analogous to the
3389 # "_adjust_for_single_table_inheritance()" method in Query.
3391 if single_crit is not None:
3392 if secondaryjoin is not None:
3393 secondaryjoin = secondaryjoin & single_crit
3394 else:
3395 primaryjoin = primaryjoin & single_crit
3397 if extra_criteria:
3399 def mark_exclude_cols(
3400 elem: SupportsAnnotations, annotations: _AnnotationDict
3401 ) -> SupportsAnnotations:
3402 """note unrelated columns in the "extra criteria" as either
3403 should be adapted or not adapted, even though they are not
3404 part of our "local" or "remote" side.
3406 see #9779 for this case, as well as #11010 for a follow up
3408 """
3410 parentmapper_for_element = elem._annotations.get(
3411 "parentmapper", None
3412 )
3414 if (
3415 # NOTE: it's not clear yet if this needs to test for
3416 # parentmapper_for_element.isa(self.prop.parent). so far
3417 # we have not come up with a test.
3418 parentmapper_for_element is not self.prop.parent
3419 and (
3420 parentmapper_for_element is None
3421 or not parentmapper_for_element.isa(self.prop.mapper)
3422 )
3423 and elem not in self._secondary_lineage_set
3424 ):
3425 return _safe_annotate(elem, annotations)
3426 else:
3427 return elem
3429 extra_criteria = tuple(
3430 _deep_annotate(
3431 elem,
3432 {"should_not_adapt": True},
3433 annotate_callable=mark_exclude_cols,
3434 )
3435 for elem in extra_criteria
3436 )
3438 if secondaryjoin is not None:
3439 secondaryjoin = secondaryjoin & sql.and_(*extra_criteria)
3440 else:
3441 primaryjoin = primaryjoin & sql.and_(*extra_criteria)
3443 if aliased:
3444 if secondary is not None:
3445 secondary = secondary._anonymous_fromclause(flat=True)
3446 primary_aliasizer = ClauseAdapter(
3447 secondary,
3448 exclude_fn=_local_col_exclude,
3449 )
3450 secondary_aliasizer = ClauseAdapter(
3451 dest_selectable, equivalents=self.child_equivalents
3452 ).chain(primary_aliasizer)
3453 if source_selectable is not None:
3454 primary_aliasizer = ClauseAdapter(
3455 secondary,
3456 exclude_fn=_local_col_exclude,
3457 ).chain(
3458 ClauseAdapter(
3459 source_selectable,
3460 equivalents=self.parent_equivalents,
3461 )
3462 )
3464 secondaryjoin = secondary_aliasizer.traverse(secondaryjoin)
3465 else:
3466 primary_aliasizer = ClauseAdapter(
3467 dest_selectable,
3468 exclude_fn=_local_col_exclude,
3469 equivalents=self.child_equivalents,
3470 )
3471 if source_selectable is not None:
3472 primary_aliasizer.chain(
3473 ClauseAdapter(
3474 source_selectable,
3475 exclude_fn=_remote_col_exclude,
3476 equivalents=self.parent_equivalents,
3477 )
3478 )
3479 secondary_aliasizer = None
3481 primaryjoin = primary_aliasizer.traverse(primaryjoin)
3482 target_adapter = secondary_aliasizer or primary_aliasizer
3483 target_adapter.exclude_fn = None
3484 else:
3485 target_adapter = None
3486 return (
3487 primaryjoin,
3488 secondaryjoin,
3489 secondary,
3490 target_adapter,
3491 dest_selectable,
3492 )
3494 def create_lazy_clause(self, reverse_direction: bool = False) -> Tuple[
3495 ColumnElement[bool],
3496 Dict[str, ColumnElement[Any]],
3497 Dict[ColumnElement[Any], ColumnElement[Any]],
3498 ]:
3499 binds: Dict[ColumnElement[Any], BindParameter[Any]] = {}
3500 equated_columns: Dict[ColumnElement[Any], ColumnElement[Any]] = {}
3502 has_secondary = self.secondaryjoin is not None
3504 if has_secondary:
3505 lookup = collections.defaultdict(list)
3506 for l, r in self.local_remote_pairs:
3507 lookup[l].append((l, r))
3508 equated_columns[r] = l
3509 elif not reverse_direction:
3510 for l, r in self.local_remote_pairs:
3511 equated_columns[r] = l
3512 else:
3513 for l, r in self.local_remote_pairs:
3514 equated_columns[l] = r
3516 def col_to_bind(
3517 element: ColumnElement[Any], **kw: Any
3518 ) -> Optional[BindParameter[Any]]:
3519 if (
3520 (not reverse_direction and "local" in element._annotations)
3521 or reverse_direction
3522 and (
3523 (has_secondary and element in lookup)
3524 or (not has_secondary and "remote" in element._annotations)
3525 )
3526 ):
3527 if element not in binds:
3528 binds[element] = sql.bindparam(
3529 None, None, type_=element.type, unique=True
3530 )
3531 return binds[element]
3532 return None
3534 lazywhere = self.primaryjoin
3535 if self.secondaryjoin is None or not reverse_direction:
3536 lazywhere = visitors.replacement_traverse(
3537 lazywhere, {}, col_to_bind
3538 )
3540 if self.secondaryjoin is not None:
3541 secondaryjoin = self.secondaryjoin
3542 if reverse_direction:
3543 secondaryjoin = visitors.replacement_traverse(
3544 secondaryjoin, {}, col_to_bind
3545 )
3546 lazywhere = sql.and_(lazywhere, secondaryjoin)
3548 bind_to_col = {binds[col].key: col for col in binds}
3550 return lazywhere, bind_to_col, equated_columns
3553class _ColInAnnotations:
3554 """Serializable object that tests for names in c._annotations.
3556 TODO: does this need to be serializable anymore? can we find what the
3557 use case was for that?
3559 """
3561 __slots__ = ("names",)
3563 def __init__(self, *names: str):
3564 self.names = frozenset(names)
3566 def __call__(self, c: ClauseElement) -> bool:
3567 return bool(self.names.intersection(c._annotations))
3570_local_col_exclude = _ColInAnnotations("local", "should_not_adapt")
3571_remote_col_exclude = _ColInAnnotations("remote", "should_not_adapt")
3574class Relationship(
3575 RelationshipProperty[_T],
3576 _DeclarativeMapped[_T],
3577):
3578 """Describes an object property that holds a single item or list
3579 of items that correspond to a related database table.
3581 Public constructor is the :func:`_orm.relationship` function.
3583 .. seealso::
3585 :ref:`relationship_config_toplevel`
3587 .. versionchanged:: 2.0 Added :class:`_orm.Relationship` as a Declarative
3588 compatible subclass for :class:`_orm.RelationshipProperty`.
3590 """
3592 inherit_cache = True
3593 """:meta private:"""
3596class _RelationshipDeclared( # type: ignore[misc]
3597 Relationship[_T],
3598 WriteOnlyMapped[_T], # not compatible with Mapped[_T]
3599 DynamicMapped[_T], # not compatible with Mapped[_T]
3600):
3601 """Relationship subclass used implicitly for declarative mapping."""
3603 inherit_cache = True
3604 """:meta private:"""
3606 @classmethod
3607 def _mapper_property_name(cls) -> str:
3608 return "Relationship"