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
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
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
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 # 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
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
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
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
1872 if isinstance(arg_origin, type) and issubclass(
1873 arg_origin, typing.Mapping
1874 ):
1875 type_arg = argument.__args__[-1] # type: ignore
1876 else:
1877 type_arg = argument.__args__[0] # type: ignore
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 and self._attribute_options.dataclasses_default_factory
1917 is not self.collection_class
1918 ):
1919 raise sa_exc.ArgumentError(
1920 f"For relationship {self._format_as_string(cls, key)} using "
1921 "dataclass options, default_factory must be exactly "
1922 f"{self.collection_class}"
1923 )
1925 @util.preload_module("sqlalchemy.orm.mapper")
1926 def _setup_entity(self, __argument: Any = None, /) -> None:
1927 if "entity" in self.__dict__:
1928 return
1930 mapperlib = util.preloaded.orm_mapper
1932 if __argument:
1933 argument = __argument
1934 else:
1935 argument = self.argument
1937 resolved_argument: _ExternalEntityType[Any]
1939 if isinstance(argument, str):
1940 # we might want to cleanup clsregistry API to make this
1941 # more straightforward
1942 resolved_argument = cast(
1943 "_ExternalEntityType[Any]",
1944 self._clsregistry_resolve_name(argument)(),
1945 )
1946 elif callable(argument) and not isinstance(
1947 argument, (type, mapperlib.Mapper)
1948 ):
1949 resolved_argument = argument()
1950 else:
1951 resolved_argument = argument
1953 entity: _InternalEntityType[Any]
1955 if isinstance(resolved_argument, type):
1956 entity = class_mapper(resolved_argument, configure=False)
1957 else:
1958 try:
1959 entity = inspect(resolved_argument)
1960 except sa_exc.NoInspectionAvailable:
1961 entity = None # type: ignore
1963 if not hasattr(entity, "mapper"):
1964 raise sa_exc.ArgumentError(
1965 "relationship '%s' expects "
1966 "a class or a mapper argument (received: %s)"
1967 % (self.key, type(resolved_argument))
1968 )
1970 self.entity = entity
1971 self.target = self.entity.persist_selectable
1973 def _setup_join_conditions(self) -> None:
1974 self._join_condition = jc = _JoinCondition(
1975 parent_persist_selectable=self.parent.persist_selectable,
1976 child_persist_selectable=self.entity.persist_selectable,
1977 parent_local_selectable=self.parent.local_table,
1978 child_local_selectable=self.entity.local_table,
1979 primaryjoin=self._init_args.primaryjoin.resolved,
1980 secondary=self._init_args.secondary.resolved,
1981 secondaryjoin=self._init_args.secondaryjoin.resolved,
1982 parent_equivalents=self.parent._equivalent_columns,
1983 child_equivalents=self.mapper._equivalent_columns,
1984 consider_as_foreign_keys=self._user_defined_foreign_keys,
1985 local_remote_pairs=self.local_remote_pairs,
1986 remote_side=self.remote_side,
1987 self_referential=self._is_self_referential,
1988 prop=self,
1989 support_sync=not self.viewonly,
1990 can_be_synced_fn=self._columns_are_mapped,
1991 )
1992 self.primaryjoin = jc.primaryjoin
1993 self.secondaryjoin = jc.secondaryjoin
1994 self.secondary = jc.secondary
1995 self.direction = jc.direction
1996 self.local_remote_pairs = jc.local_remote_pairs
1997 self.remote_side = jc.remote_columns
1998 self.local_columns = jc.local_columns
1999 self.synchronize_pairs = jc.synchronize_pairs
2000 self._calculated_foreign_keys = jc.foreign_key_columns
2001 self.secondary_synchronize_pairs = jc.secondary_synchronize_pairs
2003 @property
2004 def _clsregistry_resolve_arg(
2005 self,
2006 ) -> Callable[[str, bool], _class_resolver]:
2007 return self._clsregistry_resolvers[1]
2009 @property
2010 def _clsregistry_resolve_name(
2011 self,
2012 ) -> Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]]:
2013 return self._clsregistry_resolvers[0]
2015 @util.memoized_property
2016 @util.preload_module("sqlalchemy.orm.clsregistry")
2017 def _clsregistry_resolvers(
2018 self,
2019 ) -> Tuple[
2020 Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]],
2021 Callable[[str, bool], _class_resolver],
2022 ]:
2023 _resolver = util.preloaded.orm_clsregistry._resolver
2025 return _resolver(self.parent.class_, self)
2027 @property
2028 def cascade(self) -> CascadeOptions:
2029 """Return the current cascade setting for this
2030 :class:`.RelationshipProperty`.
2031 """
2032 return self._cascade
2034 @cascade.setter
2035 def cascade(self, cascade: Union[str, CascadeOptions]) -> None:
2036 self._set_cascade(cascade)
2038 def _set_cascade(self, cascade_arg: Union[str, CascadeOptions]) -> None:
2039 cascade = CascadeOptions(cascade_arg)
2041 if self.viewonly:
2042 cascade = CascadeOptions(
2043 cascade.intersection(CascadeOptions._viewonly_cascades)
2044 )
2046 if "mapper" in self.__dict__:
2047 self._check_cascade_settings(cascade)
2048 self._cascade = cascade
2050 if self._dependency_processor:
2051 self._dependency_processor.cascade = cascade
2053 def _check_cascade_settings(self, cascade: CascadeOptions) -> None:
2054 if (
2055 cascade.delete_orphan
2056 and not self.single_parent
2057 and (self.direction is MANYTOMANY or self.direction is MANYTOONE)
2058 ):
2059 raise sa_exc.ArgumentError(
2060 "For %(direction)s relationship %(rel)s, delete-orphan "
2061 "cascade is normally "
2062 'configured only on the "one" side of a one-to-many '
2063 "relationship, "
2064 'and not on the "many" side of a many-to-one or many-to-many '
2065 "relationship. "
2066 "To force this relationship to allow a particular "
2067 '"%(relatedcls)s" object to be referenced by only '
2068 'a single "%(clsname)s" object at a time via the '
2069 "%(rel)s relationship, which "
2070 "would allow "
2071 "delete-orphan cascade to take place in this direction, set "
2072 "the single_parent=True flag."
2073 % {
2074 "rel": self,
2075 "direction": (
2076 "many-to-one"
2077 if self.direction is MANYTOONE
2078 else "many-to-many"
2079 ),
2080 "clsname": self.parent.class_.__name__,
2081 "relatedcls": self.mapper.class_.__name__,
2082 },
2083 code="bbf0",
2084 )
2086 if self.passive_deletes == "all" and (
2087 "delete" in cascade or "delete-orphan" in cascade
2088 ):
2089 raise sa_exc.ArgumentError(
2090 "On %s, can't set passive_deletes='all' in conjunction "
2091 "with 'delete' or 'delete-orphan' cascade" % self
2092 )
2094 if cascade.delete_orphan:
2095 self.mapper.primary_mapper()._delete_orphans.append(
2096 (self.key, self.parent.class_)
2097 )
2099 def _persists_for(self, mapper: Mapper[Any]) -> bool:
2100 """Return True if this property will persist values on behalf
2101 of the given mapper.
2103 """
2105 return (
2106 self.key in mapper.relationships
2107 and mapper.relationships[self.key] is self
2108 )
2110 def _columns_are_mapped(self, *cols: ColumnElement[Any]) -> bool:
2111 """Return True if all columns in the given collection are
2112 mapped by the tables referenced by this :class:`.RelationshipProperty`.
2114 """
2116 secondary = self._init_args.secondary.resolved
2117 for c in cols:
2118 if secondary is not None and secondary.c.contains_column(c):
2119 continue
2120 if not self.parent.persist_selectable.c.contains_column(
2121 c
2122 ) and not self.target.c.contains_column(c):
2123 return False
2124 return True
2126 def _generate_backref(self) -> None:
2127 """Interpret the 'backref' instruction to create a
2128 :func:`_orm.relationship` complementary to this one."""
2130 resolve_back_populates = self._init_args.back_populates.resolved
2132 if self.backref is not None and not resolve_back_populates:
2133 kwargs: Dict[str, Any]
2134 if isinstance(self.backref, str):
2135 backref_key, kwargs = self.backref, {}
2136 else:
2137 backref_key, kwargs = self.backref
2138 mapper = self.mapper.primary_mapper()
2140 if not mapper.concrete:
2141 check = set(mapper.iterate_to_root()).union(
2142 mapper.self_and_descendants
2143 )
2144 for m in check:
2145 if m.has_property(backref_key) and not m.concrete:
2146 raise sa_exc.ArgumentError(
2147 "Error creating backref "
2148 "'%s' on relationship '%s': property of that "
2149 "name exists on mapper '%s'"
2150 % (backref_key, self, m)
2151 )
2153 # determine primaryjoin/secondaryjoin for the
2154 # backref. Use the one we had, so that
2155 # a custom join doesn't have to be specified in
2156 # both directions.
2157 if self.secondary is not None:
2158 # for many to many, just switch primaryjoin/
2159 # secondaryjoin. use the annotated
2160 # pj/sj on the _join_condition.
2161 pj = kwargs.pop(
2162 "primaryjoin",
2163 self._join_condition.secondaryjoin_minus_local,
2164 )
2165 sj = kwargs.pop(
2166 "secondaryjoin",
2167 self._join_condition.primaryjoin_minus_local,
2168 )
2169 else:
2170 pj = kwargs.pop(
2171 "primaryjoin",
2172 self._join_condition.primaryjoin_reverse_remote,
2173 )
2174 sj = kwargs.pop("secondaryjoin", None)
2175 if sj:
2176 raise sa_exc.InvalidRequestError(
2177 "Can't assign 'secondaryjoin' on a backref "
2178 "against a non-secondary relationship."
2179 )
2181 foreign_keys = kwargs.pop(
2182 "foreign_keys", self._user_defined_foreign_keys
2183 )
2184 parent = self.parent.primary_mapper()
2185 kwargs.setdefault("viewonly", self.viewonly)
2186 kwargs.setdefault("post_update", self.post_update)
2187 kwargs.setdefault("passive_updates", self.passive_updates)
2188 kwargs.setdefault("sync_backref", self.sync_backref)
2189 self.back_populates = backref_key
2190 relationship = RelationshipProperty(
2191 parent,
2192 self.secondary,
2193 primaryjoin=pj,
2194 secondaryjoin=sj,
2195 foreign_keys=foreign_keys,
2196 back_populates=self.key,
2197 **kwargs,
2198 )
2199 mapper._configure_property(
2200 backref_key, relationship, warn_for_existing=True
2201 )
2203 if resolve_back_populates:
2204 if isinstance(resolve_back_populates, PropComparator):
2205 back_populates = resolve_back_populates.prop.key
2206 elif isinstance(resolve_back_populates, str):
2207 back_populates = resolve_back_populates
2208 else:
2209 # need test coverage for this case as well
2210 raise sa_exc.ArgumentError(
2211 f"Invalid back_populates value: {resolve_back_populates!r}"
2212 )
2214 self._add_reverse_property(back_populates)
2216 @util.preload_module("sqlalchemy.orm.dependency")
2217 def _post_init(self) -> None:
2218 dependency = util.preloaded.orm_dependency
2220 if self.uselist is None:
2221 self.uselist = self.direction is not MANYTOONE
2222 if not self.viewonly:
2223 self._dependency_processor = ( # type: ignore
2224 dependency._DependencyProcessor.from_relationship
2225 )(self)
2227 if (
2228 self.uselist
2229 and self._attribute_options.dataclasses_default
2230 is not _NoArg.NO_ARG
2231 ):
2232 raise sa_exc.ArgumentError(
2233 f"On relationship {self}, the dataclass default for "
2234 "relationship may only be set for "
2235 "a relationship that references a scalar value, i.e. "
2236 "many-to-one or explicitly uselist=False"
2237 )
2239 @util.memoized_property
2240 def _use_get(self) -> bool:
2241 """memoize the 'use_get' attribute of this RelationshipLoader's
2242 lazyloader."""
2244 strategy = self._lazy_strategy
2245 return strategy.use_get
2247 @util.memoized_property
2248 def _is_self_referential(self) -> bool:
2249 return self.mapper.common_parent(self.parent)
2251 def _create_joins(
2252 self,
2253 source_polymorphic: bool = False,
2254 source_selectable: Optional[FromClause] = None,
2255 dest_selectable: Optional[FromClause] = None,
2256 of_type_entity: Optional[_InternalEntityType[Any]] = None,
2257 alias_secondary: bool = False,
2258 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
2259 ) -> Tuple[
2260 ColumnElement[bool],
2261 Optional[ColumnElement[bool]],
2262 FromClause,
2263 FromClause,
2264 Optional[FromClause],
2265 Optional[ClauseAdapter],
2266 ]:
2267 aliased = False
2269 if alias_secondary and self.secondary is not None:
2270 aliased = True
2272 if source_selectable is None:
2273 if source_polymorphic and self.parent.with_polymorphic:
2274 source_selectable = self.parent._with_polymorphic_selectable
2276 if of_type_entity:
2277 dest_mapper = of_type_entity.mapper
2278 if dest_selectable is None:
2279 dest_selectable = of_type_entity.selectable
2280 aliased = True
2281 else:
2282 dest_mapper = self.mapper
2284 if dest_selectable is None:
2285 dest_selectable = self.entity.selectable
2286 if self.mapper.with_polymorphic:
2287 aliased = True
2289 if self._is_self_referential and source_selectable is None:
2290 dest_selectable = dest_selectable._anonymous_fromclause()
2291 aliased = True
2292 elif (
2293 dest_selectable is not self.mapper._with_polymorphic_selectable
2294 or self.mapper.with_polymorphic
2295 ):
2296 aliased = True
2298 single_crit = dest_mapper._single_table_criterion
2299 aliased = aliased or (
2300 source_selectable is not None
2301 and (
2302 source_selectable
2303 is not self.parent._with_polymorphic_selectable
2304 or source_selectable._is_subquery
2305 )
2306 )
2308 (
2309 primaryjoin,
2310 secondaryjoin,
2311 secondary,
2312 target_adapter,
2313 dest_selectable,
2314 ) = self._join_condition.join_targets(
2315 source_selectable,
2316 dest_selectable,
2317 aliased,
2318 single_crit,
2319 extra_criteria,
2320 )
2321 if source_selectable is None:
2322 source_selectable = self.parent.local_table
2323 if dest_selectable is None:
2324 dest_selectable = self.entity.local_table
2325 return (
2326 primaryjoin,
2327 secondaryjoin,
2328 source_selectable,
2329 dest_selectable,
2330 secondary,
2331 target_adapter,
2332 )
2335def _annotate_columns(element: _CE, annotations: _AnnotationDict) -> _CE:
2336 def clone(elem: _CE) -> _CE:
2337 if isinstance(elem, expression.ColumnClause):
2338 elem = elem._annotate(annotations.copy()) # type: ignore
2339 elem._copy_internals(clone=clone)
2340 return elem
2342 if element is not None:
2343 element = clone(element)
2344 clone = None # type: ignore # remove gc cycles
2345 return element
2348class _JoinCondition:
2349 primaryjoin_initial: Optional[ColumnElement[bool]]
2350 primaryjoin: ColumnElement[bool]
2351 secondaryjoin: Optional[ColumnElement[bool]]
2352 secondary: Optional[FromClause]
2353 prop: RelationshipProperty[Any]
2355 synchronize_pairs: _ColumnPairs
2356 secondary_synchronize_pairs: _ColumnPairs
2357 direction: RelationshipDirection
2359 parent_persist_selectable: FromClause
2360 child_persist_selectable: FromClause
2361 parent_local_selectable: FromClause
2362 child_local_selectable: FromClause
2364 _local_remote_pairs: Optional[_ColumnPairs]
2366 def __init__(
2367 self,
2368 parent_persist_selectable: FromClause,
2369 child_persist_selectable: FromClause,
2370 parent_local_selectable: FromClause,
2371 child_local_selectable: FromClause,
2372 *,
2373 primaryjoin: Optional[ColumnElement[bool]] = None,
2374 secondary: Optional[FromClause] = None,
2375 secondaryjoin: Optional[ColumnElement[bool]] = None,
2376 parent_equivalents: Optional[_EquivalentColumnMap] = None,
2377 child_equivalents: Optional[_EquivalentColumnMap] = None,
2378 consider_as_foreign_keys: Any = None,
2379 local_remote_pairs: Optional[_ColumnPairs] = None,
2380 remote_side: Any = None,
2381 self_referential: Any = False,
2382 prop: RelationshipProperty[Any],
2383 support_sync: bool = True,
2384 can_be_synced_fn: Callable[..., bool] = lambda *c: True,
2385 ):
2386 self.parent_persist_selectable = parent_persist_selectable
2387 self.parent_local_selectable = parent_local_selectable
2388 self.child_persist_selectable = child_persist_selectable
2389 self.child_local_selectable = child_local_selectable
2390 self.parent_equivalents = parent_equivalents
2391 self.child_equivalents = child_equivalents
2392 self.primaryjoin_initial = primaryjoin
2393 self.secondaryjoin = secondaryjoin
2394 self.secondary = secondary
2395 self.consider_as_foreign_keys = consider_as_foreign_keys
2396 self._local_remote_pairs = local_remote_pairs
2397 self._remote_side = remote_side
2398 self.prop = prop
2399 self.self_referential = self_referential
2400 self.support_sync = support_sync
2401 self.can_be_synced_fn = can_be_synced_fn
2403 self._determine_joins()
2404 assert self.primaryjoin is not None
2406 self._annotate_fks()
2407 self._annotate_remote()
2408 self._annotate_local()
2409 self._annotate_parentmapper()
2410 self._setup_pairs()
2411 self._check_foreign_cols(self.primaryjoin, True)
2412 if self.secondaryjoin is not None:
2413 self._check_foreign_cols(self.secondaryjoin, False)
2414 self._determine_direction()
2415 self._check_remote_side()
2416 self._log_joins()
2418 def _log_joins(self) -> None:
2419 log = self.prop.logger
2420 log.info("%s setup primary join %s", self.prop, self.primaryjoin)
2421 log.info("%s setup secondary join %s", self.prop, self.secondaryjoin)
2422 log.info(
2423 "%s synchronize pairs [%s]",
2424 self.prop,
2425 ",".join(
2426 "(%s => %s)" % (l, r) for (l, r) in self.synchronize_pairs
2427 ),
2428 )
2429 log.info(
2430 "%s secondary synchronize pairs [%s]",
2431 self.prop,
2432 ",".join(
2433 "(%s => %s)" % (l, r)
2434 for (l, r) in self.secondary_synchronize_pairs or []
2435 ),
2436 )
2437 log.info(
2438 "%s local/remote pairs [%s]",
2439 self.prop,
2440 ",".join(
2441 "(%s / %s)" % (l, r) for (l, r) in self.local_remote_pairs
2442 ),
2443 )
2444 log.info(
2445 "%s remote columns [%s]",
2446 self.prop,
2447 ",".join("%s" % col for col in self.remote_columns),
2448 )
2449 log.info(
2450 "%s local columns [%s]",
2451 self.prop,
2452 ",".join("%s" % col for col in self.local_columns),
2453 )
2454 log.info("%s relationship direction %s", self.prop, self.direction)
2456 def _determine_joins(self) -> None:
2457 """Determine the 'primaryjoin' and 'secondaryjoin' attributes,
2458 if not passed to the constructor already.
2460 This is based on analysis of the foreign key relationships
2461 between the parent and target mapped selectables.
2463 """
2464 if self.secondaryjoin is not None and self.secondary is None:
2465 raise sa_exc.ArgumentError(
2466 "Property %s specified with secondary "
2467 "join condition but "
2468 "no secondary argument" % self.prop
2469 )
2471 # find a join between the given mapper's mapped table and
2472 # the given table. will try the mapper's local table first
2473 # for more specificity, then if not found will try the more
2474 # general mapped table, which in the case of inheritance is
2475 # a join.
2476 try:
2477 consider_as_foreign_keys = self.consider_as_foreign_keys or None
2478 if self.secondary is not None:
2479 if self.secondaryjoin is None:
2480 self.secondaryjoin = join_condition(
2481 self.child_persist_selectable,
2482 self.secondary,
2483 a_subset=self.child_local_selectable,
2484 consider_as_foreign_keys=consider_as_foreign_keys,
2485 )
2486 if self.primaryjoin_initial is None:
2487 self.primaryjoin = join_condition(
2488 self.parent_persist_selectable,
2489 self.secondary,
2490 a_subset=self.parent_local_selectable,
2491 consider_as_foreign_keys=consider_as_foreign_keys,
2492 )
2493 else:
2494 self.primaryjoin = self.primaryjoin_initial
2495 else:
2496 if self.primaryjoin_initial is None:
2497 self.primaryjoin = join_condition(
2498 self.parent_persist_selectable,
2499 self.child_persist_selectable,
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 except sa_exc.NoForeignKeysError as nfe:
2506 if self.secondary is not None:
2507 raise sa_exc.NoForeignKeysError(
2508 "Could not determine join "
2509 "condition between parent/child tables on "
2510 "relationship %s - there are no foreign keys "
2511 "linking these tables via secondary table '%s'. "
2512 "Ensure that referencing columns are associated "
2513 "with a ForeignKey or ForeignKeyConstraint, or "
2514 "specify 'primaryjoin' and 'secondaryjoin' "
2515 "expressions." % (self.prop, self.secondary)
2516 ) from nfe
2517 else:
2518 raise sa_exc.NoForeignKeysError(
2519 "Could not determine join "
2520 "condition between parent/child tables on "
2521 "relationship %s - there are no foreign keys "
2522 "linking these tables. "
2523 "Ensure that referencing columns are associated "
2524 "with a ForeignKey or ForeignKeyConstraint, or "
2525 "specify a 'primaryjoin' expression." % self.prop
2526 ) from nfe
2527 except sa_exc.AmbiguousForeignKeysError as afe:
2528 if self.secondary is not None:
2529 raise sa_exc.AmbiguousForeignKeysError(
2530 "Could not determine join "
2531 "condition between parent/child tables on "
2532 "relationship %s - there are multiple foreign key "
2533 "paths linking the tables via secondary table '%s'. "
2534 "Specify the 'foreign_keys' "
2535 "argument, providing a list of those columns which "
2536 "should be counted as containing a foreign key "
2537 "reference from the secondary table to each of the "
2538 "parent and child tables." % (self.prop, self.secondary)
2539 ) from afe
2540 else:
2541 raise sa_exc.AmbiguousForeignKeysError(
2542 "Could not determine join "
2543 "condition between parent/child tables on "
2544 "relationship %s - there are multiple foreign key "
2545 "paths linking the tables. Specify the "
2546 "'foreign_keys' argument, providing a list of those "
2547 "columns which should be counted as containing a "
2548 "foreign key reference to the parent table." % self.prop
2549 ) from afe
2551 @property
2552 def primaryjoin_minus_local(self) -> ColumnElement[bool]:
2553 return _deep_deannotate(self.primaryjoin, values=("local", "remote"))
2555 @property
2556 def secondaryjoin_minus_local(self) -> ColumnElement[bool]:
2557 assert self.secondaryjoin is not None
2558 return _deep_deannotate(self.secondaryjoin, values=("local", "remote"))
2560 @util.memoized_property
2561 def primaryjoin_reverse_remote(self) -> ColumnElement[bool]:
2562 """Return the primaryjoin condition suitable for the
2563 "reverse" direction.
2565 If the primaryjoin was delivered here with pre-existing
2566 "remote" annotations, the local/remote annotations
2567 are reversed. Otherwise, the local/remote annotations
2568 are removed.
2570 """
2571 if self._has_remote_annotations:
2573 def replace(element: _CE, **kw: Any) -> Optional[_CE]:
2574 if "remote" in element._annotations:
2575 v = dict(element._annotations)
2576 del v["remote"]
2577 v["local"] = True
2578 return element._with_annotations(v)
2579 elif "local" in element._annotations:
2580 v = dict(element._annotations)
2581 del v["local"]
2582 v["remote"] = True
2583 return element._with_annotations(v)
2585 return None
2587 return visitors.replacement_traverse(self.primaryjoin, {}, replace)
2588 else:
2589 if self._has_foreign_annotations:
2590 # TODO: coverage
2591 return _deep_deannotate(
2592 self.primaryjoin, values=("local", "remote")
2593 )
2594 else:
2595 return _deep_deannotate(self.primaryjoin)
2597 def _has_annotation(self, clause: ClauseElement, annotation: str) -> bool:
2598 for col in visitors.iterate(clause, {}):
2599 if annotation in col._annotations:
2600 return True
2601 else:
2602 return False
2604 @util.memoized_property
2605 def _has_foreign_annotations(self) -> bool:
2606 return self._has_annotation(self.primaryjoin, "foreign")
2608 @util.memoized_property
2609 def _has_remote_annotations(self) -> bool:
2610 return self._has_annotation(self.primaryjoin, "remote")
2612 @util.memoized_property
2613 def secondary_covers_parent_primary_key(self) -> bool:
2614 """Return True if the "secondary" selectable's join to the parent
2615 table contains columns that encompass the complete primary key
2616 value of the parent.
2618 Used in optimizing the selectinload loader strategy to indicate
2619 the parent table need not be included in the query, as a complete
2620 primary key can be derived from the secondary table.
2622 """
2623 if self.secondary is None:
2624 return False
2626 secondary_synced_parent_cols = util.column_set(
2627 l for (l, _) in self.synchronize_pairs
2628 )
2629 return self.prop.parent._local_pk_cols.issubset(
2630 secondary_synced_parent_cols
2631 )
2633 def _annotate_fks(self) -> None:
2634 """Annotate the primaryjoin and secondaryjoin
2635 structures with 'foreign' annotations marking columns
2636 considered as foreign.
2638 """
2639 if self._has_foreign_annotations:
2640 return
2642 if self.consider_as_foreign_keys:
2643 self._annotate_from_fk_list()
2644 else:
2645 self._annotate_present_fks()
2647 def _annotate_from_fk_list(self) -> None:
2648 def check_fk(element: _CE, **kw: Any) -> Optional[_CE]:
2649 if element in self.consider_as_foreign_keys:
2650 return element._annotate({"foreign": True})
2651 return None
2653 self.primaryjoin = visitors.replacement_traverse(
2654 self.primaryjoin, {}, check_fk
2655 )
2656 if self.secondaryjoin is not None:
2657 self.secondaryjoin = visitors.replacement_traverse(
2658 self.secondaryjoin, {}, check_fk
2659 )
2661 def _annotate_present_fks(self) -> None:
2662 if self.secondary is not None:
2663 secondarycols = util.column_set(self.secondary.c)
2664 else:
2665 secondarycols = set()
2667 def is_foreign(
2668 a: ColumnElement[Any], b: ColumnElement[Any]
2669 ) -> Optional[ColumnElement[Any]]:
2670 if isinstance(a, schema.Column) and isinstance(b, schema.Column):
2671 if a.references(b):
2672 return a
2673 elif b.references(a):
2674 return b
2676 if secondarycols:
2677 if a in secondarycols and b not in secondarycols:
2678 return a
2679 elif b in secondarycols and a not in secondarycols:
2680 return b
2682 return None
2684 def visit_binary(binary: BinaryExpression[Any]) -> None:
2685 if not isinstance(
2686 binary.left, sql.ColumnElement
2687 ) or not isinstance(binary.right, sql.ColumnElement):
2688 return
2690 if (
2691 "foreign" not in binary.left._annotations
2692 and "foreign" not in binary.right._annotations
2693 ):
2694 col = is_foreign(binary.left, binary.right)
2695 if col is not None:
2696 if col.compare(binary.left):
2697 binary.left = binary.left._annotate({"foreign": True})
2698 elif col.compare(binary.right):
2699 binary.right = binary.right._annotate(
2700 {"foreign": True}
2701 )
2703 self.primaryjoin = visitors.cloned_traverse(
2704 self.primaryjoin, {}, {"binary": visit_binary}
2705 )
2706 if self.secondaryjoin is not None:
2707 self.secondaryjoin = visitors.cloned_traverse(
2708 self.secondaryjoin, {}, {"binary": visit_binary}
2709 )
2711 def _refers_to_parent_table(self) -> bool:
2712 """Return True if the join condition contains column
2713 comparisons where both columns are in both tables.
2715 """
2716 pt = self.parent_persist_selectable
2717 mt = self.child_persist_selectable
2718 result = False
2720 def visit_binary(binary: BinaryExpression[Any]) -> None:
2721 nonlocal result
2722 c, f = binary.left, binary.right
2723 if (
2724 isinstance(c, expression.ColumnClause)
2725 and isinstance(f, expression.ColumnClause)
2726 and pt.is_derived_from(c.table)
2727 and pt.is_derived_from(f.table)
2728 and mt.is_derived_from(c.table)
2729 and mt.is_derived_from(f.table)
2730 ):
2731 result = True
2733 visitors.traverse(self.primaryjoin, {}, {"binary": visit_binary})
2734 return result
2736 def _tables_overlap(self) -> bool:
2737 """Return True if parent/child tables have some overlap."""
2739 return selectables_overlap(
2740 self.parent_persist_selectable, self.child_persist_selectable
2741 )
2743 def _annotate_remote(self) -> None:
2744 """Annotate the primaryjoin and secondaryjoin
2745 structures with 'remote' annotations marking columns
2746 considered as part of the 'remote' side.
2748 """
2749 if self._has_remote_annotations:
2750 return
2752 if self.secondary is not None:
2753 self._annotate_remote_secondary()
2754 elif self._local_remote_pairs or self._remote_side:
2755 self._annotate_remote_from_args()
2756 elif self._refers_to_parent_table():
2757 self._annotate_selfref(
2758 lambda col: "foreign" in col._annotations, False
2759 )
2760 elif self._tables_overlap():
2761 self._annotate_remote_with_overlap()
2762 else:
2763 self._annotate_remote_distinct_selectables()
2765 def _annotate_remote_secondary(self) -> None:
2766 """annotate 'remote' in primaryjoin, secondaryjoin
2767 when 'secondary' is present.
2769 """
2771 assert self.secondary is not None
2772 fixed_secondary = self.secondary
2774 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2775 if fixed_secondary.c.contains_column(element):
2776 return element._annotate({"remote": True})
2777 return None
2779 self.primaryjoin = visitors.replacement_traverse(
2780 self.primaryjoin, {}, repl
2781 )
2783 assert self.secondaryjoin is not None
2784 self.secondaryjoin = visitors.replacement_traverse(
2785 self.secondaryjoin, {}, repl
2786 )
2788 def _annotate_selfref(
2789 self, fn: Callable[[ColumnElement[Any]], bool], remote_side_given: bool
2790 ) -> None:
2791 """annotate 'remote' in primaryjoin, secondaryjoin
2792 when the relationship is detected as self-referential.
2794 """
2796 def visit_binary(binary: BinaryExpression[Any]) -> None:
2797 equated = binary.left.compare(binary.right)
2798 if isinstance(binary.left, expression.ColumnClause) and isinstance(
2799 binary.right, expression.ColumnClause
2800 ):
2801 # assume one to many - FKs are "remote"
2802 if fn(binary.left):
2803 binary.left = binary.left._annotate({"remote": True})
2804 if fn(binary.right) and not equated:
2805 binary.right = binary.right._annotate({"remote": True})
2806 elif not remote_side_given:
2807 self._warn_non_column_elements()
2809 self.primaryjoin = visitors.cloned_traverse(
2810 self.primaryjoin, {}, {"binary": visit_binary}
2811 )
2813 def _annotate_remote_from_args(self) -> None:
2814 """annotate 'remote' in primaryjoin, secondaryjoin
2815 when the 'remote_side' or '_local_remote_pairs'
2816 arguments are used.
2818 """
2819 if self._local_remote_pairs:
2820 if self._remote_side:
2821 raise sa_exc.ArgumentError(
2822 "remote_side argument is redundant "
2823 "against more detailed _local_remote_side "
2824 "argument."
2825 )
2827 remote_side = [r for (l, r) in self._local_remote_pairs]
2828 else:
2829 remote_side = self._remote_side
2831 if self._refers_to_parent_table():
2832 self._annotate_selfref(lambda col: col in remote_side, True)
2833 else:
2835 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2836 # use set() to avoid generating ``__eq__()`` expressions
2837 # against each element
2838 if element in set(remote_side):
2839 return element._annotate({"remote": True})
2840 return None
2842 self.primaryjoin = visitors.replacement_traverse(
2843 self.primaryjoin, {}, repl
2844 )
2846 def _annotate_remote_with_overlap(self) -> None:
2847 """annotate 'remote' in primaryjoin, secondaryjoin
2848 when the parent/child tables have some set of
2849 tables in common, though is not a fully self-referential
2850 relationship.
2852 """
2854 def visit_binary(binary: BinaryExpression[Any]) -> None:
2855 binary.left, binary.right = proc_left_right(
2856 binary.left, binary.right
2857 )
2858 binary.right, binary.left = proc_left_right(
2859 binary.right, binary.left
2860 )
2862 check_entities = (
2863 self.prop is not None and self.prop.mapper is not self.prop.parent
2864 )
2866 def proc_left_right(
2867 left: ColumnElement[Any], right: ColumnElement[Any]
2868 ) -> Tuple[ColumnElement[Any], ColumnElement[Any]]:
2869 if isinstance(left, expression.ColumnClause) and isinstance(
2870 right, expression.ColumnClause
2871 ):
2872 if self.child_persist_selectable.c.contains_column(
2873 right
2874 ) and self.parent_persist_selectable.c.contains_column(left):
2875 right = right._annotate({"remote": True})
2876 elif (
2877 check_entities
2878 and right._annotations.get("parentmapper") is self.prop.mapper
2879 ):
2880 right = right._annotate({"remote": True})
2881 elif (
2882 check_entities
2883 and left._annotations.get("parentmapper") is self.prop.mapper
2884 ):
2885 left = left._annotate({"remote": True})
2886 else:
2887 self._warn_non_column_elements()
2889 return left, right
2891 self.primaryjoin = visitors.cloned_traverse(
2892 self.primaryjoin, {}, {"binary": visit_binary}
2893 )
2895 def _annotate_remote_distinct_selectables(self) -> None:
2896 """annotate 'remote' in primaryjoin, secondaryjoin
2897 when the parent/child tables are entirely
2898 separate.
2900 """
2902 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2903 if self.child_persist_selectable.c.contains_column(element) and (
2904 not self.parent_local_selectable.c.contains_column(element)
2905 or self.child_local_selectable.c.contains_column(element)
2906 ):
2907 return element._annotate({"remote": True})
2908 return None
2910 self.primaryjoin = visitors.replacement_traverse(
2911 self.primaryjoin, {}, repl
2912 )
2914 def _warn_non_column_elements(self) -> None:
2915 util.warn(
2916 "Non-simple column elements in primary "
2917 "join condition for property %s - consider using "
2918 "remote() annotations to mark the remote side." % self.prop
2919 )
2921 def _annotate_local(self) -> None:
2922 """Annotate the primaryjoin and secondaryjoin
2923 structures with 'local' annotations.
2925 This annotates all column elements found
2926 simultaneously in the parent table
2927 and the join condition that don't have a
2928 'remote' annotation set up from
2929 _annotate_remote() or user-defined.
2931 """
2932 if self._has_annotation(self.primaryjoin, "local"):
2933 return
2935 if self._local_remote_pairs:
2936 local_side = util.column_set(
2937 [l for (l, r) in self._local_remote_pairs]
2938 )
2939 else:
2940 local_side = util.column_set(self.parent_persist_selectable.c)
2942 def locals_(element: _CE, **kw: Any) -> Optional[_CE]:
2943 if "remote" not in element._annotations and element in local_side:
2944 return element._annotate({"local": True})
2945 return None
2947 self.primaryjoin = visitors.replacement_traverse(
2948 self.primaryjoin, {}, locals_
2949 )
2951 def _annotate_parentmapper(self) -> None:
2952 def parentmappers_(element: _CE, **kw: Any) -> Optional[_CE]:
2953 if "remote" in element._annotations:
2954 return element._annotate({"parentmapper": self.prop.mapper})
2955 elif "local" in element._annotations:
2956 return element._annotate({"parentmapper": self.prop.parent})
2957 return None
2959 self.primaryjoin = visitors.replacement_traverse(
2960 self.primaryjoin, {}, parentmappers_
2961 )
2963 def _check_remote_side(self) -> None:
2964 if not self.local_remote_pairs:
2965 raise sa_exc.ArgumentError(
2966 "Relationship %s could "
2967 "not determine any unambiguous local/remote column "
2968 "pairs based on join condition and remote_side "
2969 "arguments. "
2970 "Consider using the remote() annotation to "
2971 "accurately mark those elements of the join "
2972 "condition that are on the remote side of "
2973 "the relationship." % (self.prop,)
2974 )
2975 else:
2976 not_target = util.column_set(
2977 self.parent_persist_selectable.c
2978 ).difference(self.child_persist_selectable.c)
2980 for _, rmt in self.local_remote_pairs:
2981 if rmt in not_target:
2982 util.warn(
2983 "Expression %s is marked as 'remote', but these "
2984 "column(s) are local to the local side. The "
2985 "remote() annotation is needed only for a "
2986 "self-referential relationship where both sides "
2987 "of the relationship refer to the same tables."
2988 % (rmt,)
2989 )
2991 def _check_foreign_cols(
2992 self, join_condition: ColumnElement[bool], primary: bool
2993 ) -> None:
2994 """Check the foreign key columns collected and emit error
2995 messages."""
2996 foreign_cols = self._gather_columns_with_annotation(
2997 join_condition, "foreign"
2998 )
3000 has_foreign = bool(foreign_cols)
3002 if primary:
3003 can_sync = bool(self.synchronize_pairs)
3004 else:
3005 can_sync = bool(self.secondary_synchronize_pairs)
3007 if (
3008 self.support_sync
3009 and can_sync
3010 or (not self.support_sync and has_foreign)
3011 ):
3012 return
3014 # from here below is just determining the best error message
3015 # to report. Check for a join condition using any operator
3016 # (not just ==), perhaps they need to turn on "viewonly=True".
3017 if self.support_sync and has_foreign and not can_sync:
3018 err = (
3019 "Could not locate any simple equality expressions "
3020 "involving locally mapped foreign key columns for "
3021 "%s join condition "
3022 "'%s' on relationship %s."
3023 % (
3024 primary and "primary" or "secondary",
3025 join_condition,
3026 self.prop,
3027 )
3028 )
3029 err += (
3030 " Ensure that referencing columns are associated "
3031 "with a ForeignKey or ForeignKeyConstraint, or are "
3032 "annotated in the join condition with the foreign() "
3033 "annotation. To allow comparison operators other than "
3034 "'==', the relationship can be marked as viewonly=True."
3035 )
3037 raise sa_exc.ArgumentError(err)
3038 else:
3039 err = (
3040 "Could not locate any relevant foreign key columns "
3041 "for %s join condition '%s' on relationship %s."
3042 % (
3043 primary and "primary" or "secondary",
3044 join_condition,
3045 self.prop,
3046 )
3047 )
3048 err += (
3049 " Ensure that referencing columns are associated "
3050 "with a ForeignKey or ForeignKeyConstraint, or are "
3051 "annotated in the join condition with the foreign() "
3052 "annotation."
3053 )
3054 raise sa_exc.ArgumentError(err)
3056 def _determine_direction(self) -> None:
3057 """Determine if this relationship is one to many, many to one,
3058 many to many.
3060 """
3061 if self.secondaryjoin is not None:
3062 self.direction = MANYTOMANY
3063 else:
3064 parentcols = util.column_set(self.parent_persist_selectable.c)
3065 targetcols = util.column_set(self.child_persist_selectable.c)
3067 # fk collection which suggests ONETOMANY.
3068 onetomany_fk = targetcols.intersection(self.foreign_key_columns)
3070 # fk collection which suggests MANYTOONE.
3072 manytoone_fk = parentcols.intersection(self.foreign_key_columns)
3074 if onetomany_fk and manytoone_fk:
3075 # fks on both sides. test for overlap of local/remote
3076 # with foreign key.
3077 # we will gather columns directly from their annotations
3078 # without deannotating, so that we can distinguish on a column
3079 # that refers to itself.
3081 # 1. columns that are both remote and FK suggest
3082 # onetomany.
3083 onetomany_local = self._gather_columns_with_annotation(
3084 self.primaryjoin, "remote", "foreign"
3085 )
3087 # 2. columns that are FK but are not remote (e.g. local)
3088 # suggest manytoone.
3089 manytoone_local = {
3090 c
3091 for c in self._gather_columns_with_annotation(
3092 self.primaryjoin, "foreign"
3093 )
3094 if "remote" not in c._annotations
3095 }
3097 # 3. if both collections are present, remove columns that
3098 # refer to themselves. This is for the case of
3099 # and_(Me.id == Me.remote_id, Me.version == Me.version)
3100 if onetomany_local and manytoone_local:
3101 self_equated = self.remote_columns.intersection(
3102 self.local_columns
3103 )
3104 onetomany_local = onetomany_local.difference(self_equated)
3105 manytoone_local = manytoone_local.difference(self_equated)
3107 # at this point, if only one or the other collection is
3108 # present, we know the direction, otherwise it's still
3109 # ambiguous.
3111 if onetomany_local and not manytoone_local:
3112 self.direction = ONETOMANY
3113 elif manytoone_local and not onetomany_local:
3114 self.direction = MANYTOONE
3115 else:
3116 raise sa_exc.ArgumentError(
3117 "Can't determine relationship"
3118 " direction for relationship '%s' - foreign "
3119 "key columns within the join condition are present "
3120 "in both the parent and the child's mapped tables. "
3121 "Ensure that only those columns referring "
3122 "to a parent column are marked as foreign, "
3123 "either via the foreign() annotation or "
3124 "via the foreign_keys argument." % self.prop
3125 )
3126 elif onetomany_fk:
3127 self.direction = ONETOMANY
3128 elif manytoone_fk:
3129 self.direction = MANYTOONE
3130 else:
3131 raise sa_exc.ArgumentError(
3132 "Can't determine relationship "
3133 "direction for relationship '%s' - foreign "
3134 "key columns are present in neither the parent "
3135 "nor the child's mapped tables" % self.prop
3136 )
3138 def _deannotate_pairs(
3139 self, collection: _ColumnPairIterable
3140 ) -> _MutableColumnPairs:
3141 """provide deannotation for the various lists of
3142 pairs, so that using them in hashes doesn't incur
3143 high-overhead __eq__() comparisons against
3144 original columns mapped.
3146 """
3147 return [(x._deannotate(), y._deannotate()) for x, y in collection]
3149 def _setup_pairs(self) -> None:
3150 sync_pairs: _MutableColumnPairs = []
3151 lrp: util.OrderedSet[Tuple[ColumnElement[Any], ColumnElement[Any]]] = (
3152 util.OrderedSet([])
3153 )
3154 secondary_sync_pairs: _MutableColumnPairs = []
3156 def go(
3157 joincond: ColumnElement[bool],
3158 collection: _MutableColumnPairs,
3159 ) -> None:
3160 def visit_binary(
3161 binary: BinaryExpression[Any],
3162 left: ColumnElement[Any],
3163 right: ColumnElement[Any],
3164 ) -> None:
3165 if (
3166 "remote" in right._annotations
3167 and "remote" not in left._annotations
3168 and self.can_be_synced_fn(left)
3169 ):
3170 lrp.add((left, right))
3171 elif (
3172 "remote" in left._annotations
3173 and "remote" not in right._annotations
3174 and self.can_be_synced_fn(right)
3175 ):
3176 lrp.add((right, left))
3177 if binary.operator is operators.eq and self.can_be_synced_fn(
3178 left, right
3179 ):
3180 if "foreign" in right._annotations:
3181 collection.append((left, right))
3182 elif "foreign" in left._annotations:
3183 collection.append((right, left))
3185 visit_binary_product(visit_binary, joincond)
3187 for joincond, collection in [
3188 (self.primaryjoin, sync_pairs),
3189 (self.secondaryjoin, secondary_sync_pairs),
3190 ]:
3191 if joincond is None:
3192 continue
3193 go(joincond, collection)
3195 self.local_remote_pairs = self._deannotate_pairs(lrp)
3196 self.synchronize_pairs = self._deannotate_pairs(sync_pairs)
3197 self.secondary_synchronize_pairs = self._deannotate_pairs(
3198 secondary_sync_pairs
3199 )
3201 _track_overlapping_sync_targets: weakref.WeakKeyDictionary[
3202 ColumnElement[Any],
3203 weakref.WeakKeyDictionary[
3204 RelationshipProperty[Any], ColumnElement[Any]
3205 ],
3206 ] = weakref.WeakKeyDictionary()
3208 def _warn_for_conflicting_sync_targets(self) -> None:
3209 if not self.support_sync:
3210 return
3212 # we would like to detect if we are synchronizing any column
3213 # pairs in conflict with another relationship that wishes to sync
3214 # an entirely different column to the same target. This is a
3215 # very rare edge case so we will try to minimize the memory/overhead
3216 # impact of this check
3217 for from_, to_ in [
3218 (from_, to_) for (from_, to_) in self.synchronize_pairs
3219 ] + [
3220 (from_, to_) for (from_, to_) in self.secondary_synchronize_pairs
3221 ]:
3222 # save ourselves a ton of memory and overhead by only
3223 # considering columns that are subject to a overlapping
3224 # FK constraints at the core level. This condition can arise
3225 # if multiple relationships overlap foreign() directly, but
3226 # we're going to assume it's typically a ForeignKeyConstraint-
3227 # level configuration that benefits from this warning.
3229 if to_ not in self._track_overlapping_sync_targets:
3230 self._track_overlapping_sync_targets[to_] = (
3231 weakref.WeakKeyDictionary({self.prop: from_})
3232 )
3233 else:
3234 other_props = []
3235 prop_to_from = self._track_overlapping_sync_targets[to_]
3237 for pr, fr_ in prop_to_from.items():
3238 if (
3239 not pr.mapper._dispose_called
3240 and pr not in self.prop._reverse_property
3241 and pr.key not in self.prop._overlaps
3242 and self.prop.key not in pr._overlaps
3243 # note: the "__*" symbol is used internally by
3244 # SQLAlchemy as a general means of suppressing the
3245 # overlaps warning for some extension cases, however
3246 # this is not currently
3247 # a publicly supported symbol and may change at
3248 # any time.
3249 and "__*" not in self.prop._overlaps
3250 and "__*" not in pr._overlaps
3251 and not self.prop.parent.is_sibling(pr.parent)
3252 and not self.prop.mapper.is_sibling(pr.mapper)
3253 and not self.prop.parent.is_sibling(pr.mapper)
3254 and not self.prop.mapper.is_sibling(pr.parent)
3255 and (
3256 self.prop.key != pr.key
3257 or not self.prop.parent.common_parent(pr.parent)
3258 )
3259 ):
3260 other_props.append((pr, fr_))
3262 if other_props:
3263 util.warn(
3264 "relationship '%s' will copy column %s to column %s, "
3265 "which conflicts with relationship(s): %s. "
3266 "If this is not the intention, consider if these "
3267 "relationships should be linked with "
3268 "back_populates, or if viewonly=True should be "
3269 "applied to one or more if they are read-only. "
3270 "For the less common case that foreign key "
3271 "constraints are partially overlapping, the "
3272 "orm.foreign() "
3273 "annotation can be used to isolate the columns that "
3274 "should be written towards. To silence this "
3275 "warning, add the parameter 'overlaps=\"%s\"' to the "
3276 "'%s' relationship."
3277 % (
3278 self.prop,
3279 from_,
3280 to_,
3281 ", ".join(
3282 sorted(
3283 "'%s' (copies %s to %s)" % (pr, fr_, to_)
3284 for (pr, fr_) in other_props
3285 )
3286 ),
3287 ",".join(sorted(pr.key for pr, fr in other_props)),
3288 self.prop,
3289 ),
3290 code="qzyx",
3291 )
3292 self._track_overlapping_sync_targets[to_][self.prop] = from_
3294 @util.memoized_property
3295 def remote_columns(self) -> Set[ColumnElement[Any]]:
3296 return self._gather_join_annotations("remote")
3298 @util.memoized_property
3299 def local_columns(self) -> Set[ColumnElement[Any]]:
3300 return self._gather_join_annotations("local")
3302 @util.memoized_property
3303 def foreign_key_columns(self) -> Set[ColumnElement[Any]]:
3304 return self._gather_join_annotations("foreign")
3306 def _gather_join_annotations(
3307 self, annotation: str
3308 ) -> Set[ColumnElement[Any]]:
3309 s = set(
3310 self._gather_columns_with_annotation(self.primaryjoin, annotation)
3311 )
3312 if self.secondaryjoin is not None:
3313 s.update(
3314 self._gather_columns_with_annotation(
3315 self.secondaryjoin, annotation
3316 )
3317 )
3318 return {x._deannotate() for x in s}
3320 def _gather_columns_with_annotation(
3321 self, clause: ColumnElement[Any], *annotation: Iterable[str]
3322 ) -> Set[ColumnElement[Any]]:
3323 annotation_set = set(annotation)
3324 return {
3325 cast(ColumnElement[Any], col)
3326 for col in visitors.iterate(clause, {})
3327 if annotation_set.issubset(col._annotations)
3328 }
3330 @util.memoized_property
3331 def _secondary_lineage_set(self) -> FrozenSet[ColumnElement[Any]]:
3332 if self.secondary is not None:
3333 return frozenset(
3334 itertools.chain(*[c.proxy_set for c in self.secondary.c])
3335 )
3336 else:
3337 return util.EMPTY_SET
3339 def join_targets(
3340 self,
3341 source_selectable: Optional[FromClause],
3342 dest_selectable: FromClause,
3343 aliased: bool,
3344 single_crit: Optional[ColumnElement[bool]] = None,
3345 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
3346 ) -> Tuple[
3347 ColumnElement[bool],
3348 Optional[ColumnElement[bool]],
3349 Optional[FromClause],
3350 Optional[ClauseAdapter],
3351 FromClause,
3352 ]:
3353 """Given a source and destination selectable, create a
3354 join between them.
3356 This takes into account aliasing the join clause
3357 to reference the appropriate corresponding columns
3358 in the target objects, as well as the extra child
3359 criterion, equivalent column sets, etc.
3361 """
3362 # place a barrier on the destination such that
3363 # replacement traversals won't ever dig into it.
3364 # its internal structure remains fixed
3365 # regardless of context.
3366 dest_selectable = _shallow_annotate(
3367 dest_selectable, {"no_replacement_traverse": True}
3368 )
3370 primaryjoin, secondaryjoin, secondary = (
3371 self.primaryjoin,
3372 self.secondaryjoin,
3373 self.secondary,
3374 )
3376 # adjust the join condition for single table inheritance,
3377 # in the case that the join is to a subclass
3378 # this is analogous to the
3379 # "_adjust_for_single_table_inheritance()" method in Query.
3381 if single_crit is not None:
3382 if secondaryjoin is not None:
3383 secondaryjoin = secondaryjoin & single_crit
3384 else:
3385 primaryjoin = primaryjoin & single_crit
3387 if extra_criteria:
3389 def mark_exclude_cols(
3390 elem: SupportsAnnotations, annotations: _AnnotationDict
3391 ) -> SupportsAnnotations:
3392 """note unrelated columns in the "extra criteria" as either
3393 should be adapted or not adapted, even though they are not
3394 part of our "local" or "remote" side.
3396 see #9779 for this case, as well as #11010 for a follow up
3398 """
3400 parentmapper_for_element = elem._annotations.get(
3401 "parentmapper", None
3402 )
3404 if (
3405 # NOTE: it's not clear yet if this needs to test for
3406 # parentmapper_for_element.isa(self.prop.parent). so far
3407 # we have not come up with a test.
3408 parentmapper_for_element is not self.prop.parent
3409 and (
3410 parentmapper_for_element is None
3411 or not parentmapper_for_element.isa(self.prop.mapper)
3412 )
3413 and elem not in self._secondary_lineage_set
3414 ):
3415 return _safe_annotate(elem, annotations)
3416 else:
3417 return elem
3419 extra_criteria = tuple(
3420 _deep_annotate(
3421 elem,
3422 {"should_not_adapt": True},
3423 annotate_callable=mark_exclude_cols,
3424 )
3425 for elem in extra_criteria
3426 )
3428 if secondaryjoin is not None:
3429 secondaryjoin = secondaryjoin & sql.and_(*extra_criteria)
3430 else:
3431 primaryjoin = primaryjoin & sql.and_(*extra_criteria)
3433 if aliased:
3434 if secondary is not None:
3435 secondary = secondary._anonymous_fromclause(flat=True)
3436 primary_aliasizer = ClauseAdapter(
3437 secondary,
3438 exclude_fn=_local_col_exclude,
3439 )
3440 secondary_aliasizer = ClauseAdapter(
3441 dest_selectable, equivalents=self.child_equivalents
3442 ).chain(primary_aliasizer)
3443 if source_selectable is not None:
3444 primary_aliasizer = ClauseAdapter(
3445 secondary,
3446 exclude_fn=_local_col_exclude,
3447 ).chain(
3448 ClauseAdapter(
3449 source_selectable,
3450 equivalents=self.parent_equivalents,
3451 )
3452 )
3454 secondaryjoin = secondary_aliasizer.traverse(secondaryjoin)
3455 else:
3456 primary_aliasizer = ClauseAdapter(
3457 dest_selectable,
3458 exclude_fn=_local_col_exclude,
3459 equivalents=self.child_equivalents,
3460 )
3461 if source_selectable is not None:
3462 primary_aliasizer.chain(
3463 ClauseAdapter(
3464 source_selectable,
3465 exclude_fn=_remote_col_exclude,
3466 equivalents=self.parent_equivalents,
3467 )
3468 )
3469 secondary_aliasizer = None
3471 primaryjoin = primary_aliasizer.traverse(primaryjoin)
3472 target_adapter = secondary_aliasizer or primary_aliasizer
3473 target_adapter.exclude_fn = None
3474 else:
3475 target_adapter = None
3476 return (
3477 primaryjoin,
3478 secondaryjoin,
3479 secondary,
3480 target_adapter,
3481 dest_selectable,
3482 )
3484 def create_lazy_clause(self, reverse_direction: bool = False) -> Tuple[
3485 ColumnElement[bool],
3486 Dict[str, ColumnElement[Any]],
3487 Dict[ColumnElement[Any], ColumnElement[Any]],
3488 ]:
3489 binds: Dict[ColumnElement[Any], BindParameter[Any]] = {}
3490 equated_columns: Dict[ColumnElement[Any], ColumnElement[Any]] = {}
3492 has_secondary = self.secondaryjoin is not None
3494 if has_secondary:
3495 lookup = collections.defaultdict(list)
3496 for l, r in self.local_remote_pairs:
3497 lookup[l].append((l, r))
3498 equated_columns[r] = l
3499 elif not reverse_direction:
3500 for l, r in self.local_remote_pairs:
3501 equated_columns[r] = l
3502 else:
3503 for l, r in self.local_remote_pairs:
3504 equated_columns[l] = r
3506 def col_to_bind(
3507 element: ColumnElement[Any], **kw: Any
3508 ) -> Optional[BindParameter[Any]]:
3509 if (
3510 (not reverse_direction and "local" in element._annotations)
3511 or reverse_direction
3512 and (
3513 (has_secondary and element in lookup)
3514 or (not has_secondary and "remote" in element._annotations)
3515 )
3516 ):
3517 if element not in binds:
3518 binds[element] = sql.bindparam(
3519 None, None, type_=element.type, unique=True
3520 )
3521 return binds[element]
3522 return None
3524 lazywhere = self.primaryjoin
3525 if self.secondaryjoin is None or not reverse_direction:
3526 lazywhere = visitors.replacement_traverse(
3527 lazywhere, {}, col_to_bind
3528 )
3530 if self.secondaryjoin is not None:
3531 secondaryjoin = self.secondaryjoin
3532 if reverse_direction:
3533 secondaryjoin = visitors.replacement_traverse(
3534 secondaryjoin, {}, col_to_bind
3535 )
3536 lazywhere = sql.and_(lazywhere, secondaryjoin)
3538 bind_to_col = {binds[col].key: col for col in binds}
3540 return lazywhere, bind_to_col, equated_columns
3543class _ColInAnnotations:
3544 """Serializable object that tests for names in c._annotations.
3546 TODO: does this need to be serializable anymore? can we find what the
3547 use case was for that?
3549 """
3551 __slots__ = ("names",)
3553 def __init__(self, *names: str):
3554 self.names = frozenset(names)
3556 def __call__(self, c: ClauseElement) -> bool:
3557 return bool(self.names.intersection(c._annotations))
3560_local_col_exclude = _ColInAnnotations("local", "should_not_adapt")
3561_remote_col_exclude = _ColInAnnotations("remote", "should_not_adapt")
3564class Relationship(
3565 RelationshipProperty[_T],
3566 _DeclarativeMapped[_T],
3567):
3568 """Describes an object property that holds a single item or list
3569 of items that correspond to a related database table.
3571 Public constructor is the :func:`_orm.relationship` function.
3573 .. seealso::
3575 :ref:`relationship_config_toplevel`
3577 .. versionchanged:: 2.0 Added :class:`_orm.Relationship` as a Declarative
3578 compatible subclass for :class:`_orm.RelationshipProperty`.
3580 """
3582 inherit_cache = True
3583 """:meta private:"""
3586class _RelationshipDeclared( # type: ignore[misc]
3587 Relationship[_T],
3588 WriteOnlyMapped[_T], # not compatible with Mapped[_T]
3589 DynamicMapped[_T], # not compatible with Mapped[_T]
3590):
3591 """Relationship subclass used implicitly for declarative mapping."""
3593 inherit_cache = True
3594 """:meta private:"""
3596 @classmethod
3597 def _mapper_property_name(cls) -> str:
3598 return "Relationship"