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 def _annotate_fks(self) -> None:
2613 """Annotate the primaryjoin and secondaryjoin
2614 structures with 'foreign' annotations marking columns
2615 considered as foreign.
2617 """
2618 if self._has_foreign_annotations:
2619 return
2621 if self.consider_as_foreign_keys:
2622 self._annotate_from_fk_list()
2623 else:
2624 self._annotate_present_fks()
2626 def _annotate_from_fk_list(self) -> None:
2627 def check_fk(element: _CE, **kw: Any) -> Optional[_CE]:
2628 if element in self.consider_as_foreign_keys:
2629 return element._annotate({"foreign": True})
2630 return None
2632 self.primaryjoin = visitors.replacement_traverse(
2633 self.primaryjoin, {}, check_fk
2634 )
2635 if self.secondaryjoin is not None:
2636 self.secondaryjoin = visitors.replacement_traverse(
2637 self.secondaryjoin, {}, check_fk
2638 )
2640 def _annotate_present_fks(self) -> None:
2641 if self.secondary is not None:
2642 secondarycols = util.column_set(self.secondary.c)
2643 else:
2644 secondarycols = set()
2646 def is_foreign(
2647 a: ColumnElement[Any], b: ColumnElement[Any]
2648 ) -> Optional[ColumnElement[Any]]:
2649 if isinstance(a, schema.Column) and isinstance(b, schema.Column):
2650 if a.references(b):
2651 return a
2652 elif b.references(a):
2653 return b
2655 if secondarycols:
2656 if a in secondarycols and b not in secondarycols:
2657 return a
2658 elif b in secondarycols and a not in secondarycols:
2659 return b
2661 return None
2663 def visit_binary(binary: BinaryExpression[Any]) -> None:
2664 if not isinstance(
2665 binary.left, sql.ColumnElement
2666 ) or not isinstance(binary.right, sql.ColumnElement):
2667 return
2669 if (
2670 "foreign" not in binary.left._annotations
2671 and "foreign" not in binary.right._annotations
2672 ):
2673 col = is_foreign(binary.left, binary.right)
2674 if col is not None:
2675 if col.compare(binary.left):
2676 binary.left = binary.left._annotate({"foreign": True})
2677 elif col.compare(binary.right):
2678 binary.right = binary.right._annotate(
2679 {"foreign": True}
2680 )
2682 self.primaryjoin = visitors.cloned_traverse(
2683 self.primaryjoin, {}, {"binary": visit_binary}
2684 )
2685 if self.secondaryjoin is not None:
2686 self.secondaryjoin = visitors.cloned_traverse(
2687 self.secondaryjoin, {}, {"binary": visit_binary}
2688 )
2690 def _refers_to_parent_table(self) -> bool:
2691 """Return True if the join condition contains column
2692 comparisons where both columns are in both tables.
2694 """
2695 pt = self.parent_persist_selectable
2696 mt = self.child_persist_selectable
2697 result = False
2699 def visit_binary(binary: BinaryExpression[Any]) -> None:
2700 nonlocal result
2701 c, f = binary.left, binary.right
2702 if (
2703 isinstance(c, expression.ColumnClause)
2704 and isinstance(f, expression.ColumnClause)
2705 and pt.is_derived_from(c.table)
2706 and pt.is_derived_from(f.table)
2707 and mt.is_derived_from(c.table)
2708 and mt.is_derived_from(f.table)
2709 ):
2710 result = True
2712 visitors.traverse(self.primaryjoin, {}, {"binary": visit_binary})
2713 return result
2715 def _tables_overlap(self) -> bool:
2716 """Return True if parent/child tables have some overlap."""
2718 return selectables_overlap(
2719 self.parent_persist_selectable, self.child_persist_selectable
2720 )
2722 def _annotate_remote(self) -> None:
2723 """Annotate the primaryjoin and secondaryjoin
2724 structures with 'remote' annotations marking columns
2725 considered as part of the 'remote' side.
2727 """
2728 if self._has_remote_annotations:
2729 return
2731 if self.secondary is not None:
2732 self._annotate_remote_secondary()
2733 elif self._local_remote_pairs or self._remote_side:
2734 self._annotate_remote_from_args()
2735 elif self._refers_to_parent_table():
2736 self._annotate_selfref(
2737 lambda col: "foreign" in col._annotations, False
2738 )
2739 elif self._tables_overlap():
2740 self._annotate_remote_with_overlap()
2741 else:
2742 self._annotate_remote_distinct_selectables()
2744 def _annotate_remote_secondary(self) -> None:
2745 """annotate 'remote' in primaryjoin, secondaryjoin
2746 when 'secondary' is present.
2748 """
2750 assert self.secondary is not None
2751 fixed_secondary = self.secondary
2753 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2754 if fixed_secondary.c.contains_column(element):
2755 return element._annotate({"remote": True})
2756 return None
2758 self.primaryjoin = visitors.replacement_traverse(
2759 self.primaryjoin, {}, repl
2760 )
2762 assert self.secondaryjoin is not None
2763 self.secondaryjoin = visitors.replacement_traverse(
2764 self.secondaryjoin, {}, repl
2765 )
2767 def _annotate_selfref(
2768 self, fn: Callable[[ColumnElement[Any]], bool], remote_side_given: bool
2769 ) -> None:
2770 """annotate 'remote' in primaryjoin, secondaryjoin
2771 when the relationship is detected as self-referential.
2773 """
2775 def visit_binary(binary: BinaryExpression[Any]) -> None:
2776 equated = binary.left.compare(binary.right)
2777 if isinstance(binary.left, expression.ColumnClause) and isinstance(
2778 binary.right, expression.ColumnClause
2779 ):
2780 # assume one to many - FKs are "remote"
2781 if fn(binary.left):
2782 binary.left = binary.left._annotate({"remote": True})
2783 if fn(binary.right) and not equated:
2784 binary.right = binary.right._annotate({"remote": True})
2785 elif not remote_side_given:
2786 self._warn_non_column_elements()
2788 self.primaryjoin = visitors.cloned_traverse(
2789 self.primaryjoin, {}, {"binary": visit_binary}
2790 )
2792 def _annotate_remote_from_args(self) -> None:
2793 """annotate 'remote' in primaryjoin, secondaryjoin
2794 when the 'remote_side' or '_local_remote_pairs'
2795 arguments are used.
2797 """
2798 if self._local_remote_pairs:
2799 if self._remote_side:
2800 raise sa_exc.ArgumentError(
2801 "remote_side argument is redundant "
2802 "against more detailed _local_remote_side "
2803 "argument."
2804 )
2806 remote_side = [r for (l, r) in self._local_remote_pairs]
2807 else:
2808 remote_side = self._remote_side
2810 if self._refers_to_parent_table():
2811 self._annotate_selfref(lambda col: col in remote_side, True)
2812 else:
2814 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2815 # use set() to avoid generating ``__eq__()`` expressions
2816 # against each element
2817 if element in set(remote_side):
2818 return element._annotate({"remote": True})
2819 return None
2821 self.primaryjoin = visitors.replacement_traverse(
2822 self.primaryjoin, {}, repl
2823 )
2825 def _annotate_remote_with_overlap(self) -> None:
2826 """annotate 'remote' in primaryjoin, secondaryjoin
2827 when the parent/child tables have some set of
2828 tables in common, though is not a fully self-referential
2829 relationship.
2831 """
2833 def visit_binary(binary: BinaryExpression[Any]) -> None:
2834 binary.left, binary.right = proc_left_right(
2835 binary.left, binary.right
2836 )
2837 binary.right, binary.left = proc_left_right(
2838 binary.right, binary.left
2839 )
2841 check_entities = (
2842 self.prop is not None and self.prop.mapper is not self.prop.parent
2843 )
2845 def proc_left_right(
2846 left: ColumnElement[Any], right: ColumnElement[Any]
2847 ) -> Tuple[ColumnElement[Any], ColumnElement[Any]]:
2848 if isinstance(left, expression.ColumnClause) and isinstance(
2849 right, expression.ColumnClause
2850 ):
2851 if self.child_persist_selectable.c.contains_column(
2852 right
2853 ) and self.parent_persist_selectable.c.contains_column(left):
2854 right = right._annotate({"remote": True})
2855 elif (
2856 check_entities
2857 and right._annotations.get("parentmapper") is self.prop.mapper
2858 ):
2859 right = right._annotate({"remote": True})
2860 elif (
2861 check_entities
2862 and left._annotations.get("parentmapper") is self.prop.mapper
2863 ):
2864 left = left._annotate({"remote": True})
2865 else:
2866 self._warn_non_column_elements()
2868 return left, right
2870 self.primaryjoin = visitors.cloned_traverse(
2871 self.primaryjoin, {}, {"binary": visit_binary}
2872 )
2874 def _annotate_remote_distinct_selectables(self) -> None:
2875 """annotate 'remote' in primaryjoin, secondaryjoin
2876 when the parent/child tables are entirely
2877 separate.
2879 """
2881 def repl(element: _CE, **kw: Any) -> Optional[_CE]:
2882 if self.child_persist_selectable.c.contains_column(element) and (
2883 not self.parent_local_selectable.c.contains_column(element)
2884 or self.child_local_selectable.c.contains_column(element)
2885 ):
2886 return element._annotate({"remote": True})
2887 return None
2889 self.primaryjoin = visitors.replacement_traverse(
2890 self.primaryjoin, {}, repl
2891 )
2893 def _warn_non_column_elements(self) -> None:
2894 util.warn(
2895 "Non-simple column elements in primary "
2896 "join condition for property %s - consider using "
2897 "remote() annotations to mark the remote side." % self.prop
2898 )
2900 def _annotate_local(self) -> None:
2901 """Annotate the primaryjoin and secondaryjoin
2902 structures with 'local' annotations.
2904 This annotates all column elements found
2905 simultaneously in the parent table
2906 and the join condition that don't have a
2907 'remote' annotation set up from
2908 _annotate_remote() or user-defined.
2910 """
2911 if self._has_annotation(self.primaryjoin, "local"):
2912 return
2914 if self._local_remote_pairs:
2915 local_side = util.column_set(
2916 [l for (l, r) in self._local_remote_pairs]
2917 )
2918 else:
2919 local_side = util.column_set(self.parent_persist_selectable.c)
2921 def locals_(element: _CE, **kw: Any) -> Optional[_CE]:
2922 if "remote" not in element._annotations and element in local_side:
2923 return element._annotate({"local": True})
2924 return None
2926 self.primaryjoin = visitors.replacement_traverse(
2927 self.primaryjoin, {}, locals_
2928 )
2930 def _annotate_parentmapper(self) -> None:
2931 def parentmappers_(element: _CE, **kw: Any) -> Optional[_CE]:
2932 if "remote" in element._annotations:
2933 return element._annotate({"parentmapper": self.prop.mapper})
2934 elif "local" in element._annotations:
2935 return element._annotate({"parentmapper": self.prop.parent})
2936 return None
2938 self.primaryjoin = visitors.replacement_traverse(
2939 self.primaryjoin, {}, parentmappers_
2940 )
2942 def _check_remote_side(self) -> None:
2943 if not self.local_remote_pairs:
2944 raise sa_exc.ArgumentError(
2945 "Relationship %s could "
2946 "not determine any unambiguous local/remote column "
2947 "pairs based on join condition and remote_side "
2948 "arguments. "
2949 "Consider using the remote() annotation to "
2950 "accurately mark those elements of the join "
2951 "condition that are on the remote side of "
2952 "the relationship." % (self.prop,)
2953 )
2954 else:
2955 not_target = util.column_set(
2956 self.parent_persist_selectable.c
2957 ).difference(self.child_persist_selectable.c)
2959 for _, rmt in self.local_remote_pairs:
2960 if rmt in not_target:
2961 util.warn(
2962 "Expression %s is marked as 'remote', but these "
2963 "column(s) are local to the local side. The "
2964 "remote() annotation is needed only for a "
2965 "self-referential relationship where both sides "
2966 "of the relationship refer to the same tables."
2967 % (rmt,)
2968 )
2970 def _check_foreign_cols(
2971 self, join_condition: ColumnElement[bool], primary: bool
2972 ) -> None:
2973 """Check the foreign key columns collected and emit error
2974 messages."""
2975 foreign_cols = self._gather_columns_with_annotation(
2976 join_condition, "foreign"
2977 )
2979 has_foreign = bool(foreign_cols)
2981 if primary:
2982 can_sync = bool(self.synchronize_pairs)
2983 else:
2984 can_sync = bool(self.secondary_synchronize_pairs)
2986 if (
2987 self.support_sync
2988 and can_sync
2989 or (not self.support_sync and has_foreign)
2990 ):
2991 return
2993 # from here below is just determining the best error message
2994 # to report. Check for a join condition using any operator
2995 # (not just ==), perhaps they need to turn on "viewonly=True".
2996 if self.support_sync and has_foreign and not can_sync:
2997 err = (
2998 "Could not locate any simple equality expressions "
2999 "involving locally mapped foreign key columns for "
3000 "%s join condition "
3001 "'%s' on relationship %s."
3002 % (
3003 primary and "primary" or "secondary",
3004 join_condition,
3005 self.prop,
3006 )
3007 )
3008 err += (
3009 " Ensure that referencing columns are associated "
3010 "with a ForeignKey or ForeignKeyConstraint, or are "
3011 "annotated in the join condition with the foreign() "
3012 "annotation. To allow comparison operators other than "
3013 "'==', the relationship can be marked as viewonly=True."
3014 )
3016 raise sa_exc.ArgumentError(err)
3017 else:
3018 err = (
3019 "Could not locate any relevant foreign key columns "
3020 "for %s join condition '%s' on relationship %s."
3021 % (
3022 primary and "primary" or "secondary",
3023 join_condition,
3024 self.prop,
3025 )
3026 )
3027 err += (
3028 " Ensure that referencing columns are associated "
3029 "with a ForeignKey or ForeignKeyConstraint, or are "
3030 "annotated in the join condition with the foreign() "
3031 "annotation."
3032 )
3033 raise sa_exc.ArgumentError(err)
3035 def _determine_direction(self) -> None:
3036 """Determine if this relationship is one to many, many to one,
3037 many to many.
3039 """
3040 if self.secondaryjoin is not None:
3041 self.direction = MANYTOMANY
3042 else:
3043 parentcols = util.column_set(self.parent_persist_selectable.c)
3044 targetcols = util.column_set(self.child_persist_selectable.c)
3046 # fk collection which suggests ONETOMANY.
3047 onetomany_fk = targetcols.intersection(self.foreign_key_columns)
3049 # fk collection which suggests MANYTOONE.
3051 manytoone_fk = parentcols.intersection(self.foreign_key_columns)
3053 if onetomany_fk and manytoone_fk:
3054 # fks on both sides. test for overlap of local/remote
3055 # with foreign key.
3056 # we will gather columns directly from their annotations
3057 # without deannotating, so that we can distinguish on a column
3058 # that refers to itself.
3060 # 1. columns that are both remote and FK suggest
3061 # onetomany.
3062 onetomany_local = self._gather_columns_with_annotation(
3063 self.primaryjoin, "remote", "foreign"
3064 )
3066 # 2. columns that are FK but are not remote (e.g. local)
3067 # suggest manytoone.
3068 manytoone_local = {
3069 c
3070 for c in self._gather_columns_with_annotation(
3071 self.primaryjoin, "foreign"
3072 )
3073 if "remote" not in c._annotations
3074 }
3076 # 3. if both collections are present, remove columns that
3077 # refer to themselves. This is for the case of
3078 # and_(Me.id == Me.remote_id, Me.version == Me.version)
3079 if onetomany_local and manytoone_local:
3080 self_equated = self.remote_columns.intersection(
3081 self.local_columns
3082 )
3083 onetomany_local = onetomany_local.difference(self_equated)
3084 manytoone_local = manytoone_local.difference(self_equated)
3086 # at this point, if only one or the other collection is
3087 # present, we know the direction, otherwise it's still
3088 # ambiguous.
3090 if onetomany_local and not manytoone_local:
3091 self.direction = ONETOMANY
3092 elif manytoone_local and not onetomany_local:
3093 self.direction = MANYTOONE
3094 else:
3095 raise sa_exc.ArgumentError(
3096 "Can't determine relationship"
3097 " direction for relationship '%s' - foreign "
3098 "key columns within the join condition are present "
3099 "in both the parent and the child's mapped tables. "
3100 "Ensure that only those columns referring "
3101 "to a parent column are marked as foreign, "
3102 "either via the foreign() annotation or "
3103 "via the foreign_keys argument." % self.prop
3104 )
3105 elif onetomany_fk:
3106 self.direction = ONETOMANY
3107 elif manytoone_fk:
3108 self.direction = MANYTOONE
3109 else:
3110 raise sa_exc.ArgumentError(
3111 "Can't determine relationship "
3112 "direction for relationship '%s' - foreign "
3113 "key columns are present in neither the parent "
3114 "nor the child's mapped tables" % self.prop
3115 )
3117 def _deannotate_pairs(
3118 self, collection: _ColumnPairIterable
3119 ) -> _MutableColumnPairs:
3120 """provide deannotation for the various lists of
3121 pairs, so that using them in hashes doesn't incur
3122 high-overhead __eq__() comparisons against
3123 original columns mapped.
3125 """
3126 return [(x._deannotate(), y._deannotate()) for x, y in collection]
3128 def _setup_pairs(self) -> None:
3129 sync_pairs: _MutableColumnPairs = []
3130 lrp: util.OrderedSet[Tuple[ColumnElement[Any], ColumnElement[Any]]] = (
3131 util.OrderedSet([])
3132 )
3133 secondary_sync_pairs: _MutableColumnPairs = []
3135 def go(
3136 joincond: ColumnElement[bool],
3137 collection: _MutableColumnPairs,
3138 ) -> None:
3139 def visit_binary(
3140 binary: BinaryExpression[Any],
3141 left: ColumnElement[Any],
3142 right: ColumnElement[Any],
3143 ) -> None:
3144 if (
3145 "remote" in right._annotations
3146 and "remote" not in left._annotations
3147 and self.can_be_synced_fn(left)
3148 ):
3149 lrp.add((left, right))
3150 elif (
3151 "remote" in left._annotations
3152 and "remote" not in right._annotations
3153 and self.can_be_synced_fn(right)
3154 ):
3155 lrp.add((right, left))
3156 if binary.operator is operators.eq and self.can_be_synced_fn(
3157 left, right
3158 ):
3159 if "foreign" in right._annotations:
3160 collection.append((left, right))
3161 elif "foreign" in left._annotations:
3162 collection.append((right, left))
3164 visit_binary_product(visit_binary, joincond)
3166 for joincond, collection in [
3167 (self.primaryjoin, sync_pairs),
3168 (self.secondaryjoin, secondary_sync_pairs),
3169 ]:
3170 if joincond is None:
3171 continue
3172 go(joincond, collection)
3174 self.local_remote_pairs = self._deannotate_pairs(lrp)
3175 self.synchronize_pairs = self._deannotate_pairs(sync_pairs)
3176 self.secondary_synchronize_pairs = self._deannotate_pairs(
3177 secondary_sync_pairs
3178 )
3180 _track_overlapping_sync_targets: weakref.WeakKeyDictionary[
3181 ColumnElement[Any],
3182 weakref.WeakKeyDictionary[
3183 RelationshipProperty[Any], ColumnElement[Any]
3184 ],
3185 ] = weakref.WeakKeyDictionary()
3187 def _warn_for_conflicting_sync_targets(self) -> None:
3188 if not self.support_sync:
3189 return
3191 # we would like to detect if we are synchronizing any column
3192 # pairs in conflict with another relationship that wishes to sync
3193 # an entirely different column to the same target. This is a
3194 # very rare edge case so we will try to minimize the memory/overhead
3195 # impact of this check
3196 for from_, to_ in [
3197 (from_, to_) for (from_, to_) in self.synchronize_pairs
3198 ] + [
3199 (from_, to_) for (from_, to_) in self.secondary_synchronize_pairs
3200 ]:
3201 # save ourselves a ton of memory and overhead by only
3202 # considering columns that are subject to a overlapping
3203 # FK constraints at the core level. This condition can arise
3204 # if multiple relationships overlap foreign() directly, but
3205 # we're going to assume it's typically a ForeignKeyConstraint-
3206 # level configuration that benefits from this warning.
3208 if to_ not in self._track_overlapping_sync_targets:
3209 self._track_overlapping_sync_targets[to_] = (
3210 weakref.WeakKeyDictionary({self.prop: from_})
3211 )
3212 else:
3213 other_props = []
3214 prop_to_from = self._track_overlapping_sync_targets[to_]
3216 for pr, fr_ in prop_to_from.items():
3217 if (
3218 not pr.mapper._dispose_called
3219 and pr not in self.prop._reverse_property
3220 and pr.key not in self.prop._overlaps
3221 and self.prop.key not in pr._overlaps
3222 # note: the "__*" symbol is used internally by
3223 # SQLAlchemy as a general means of suppressing the
3224 # overlaps warning for some extension cases, however
3225 # this is not currently
3226 # a publicly supported symbol and may change at
3227 # any time.
3228 and "__*" not in self.prop._overlaps
3229 and "__*" not in pr._overlaps
3230 and not self.prop.parent.is_sibling(pr.parent)
3231 and not self.prop.mapper.is_sibling(pr.mapper)
3232 and not self.prop.parent.is_sibling(pr.mapper)
3233 and not self.prop.mapper.is_sibling(pr.parent)
3234 and (
3235 self.prop.key != pr.key
3236 or not self.prop.parent.common_parent(pr.parent)
3237 )
3238 ):
3239 other_props.append((pr, fr_))
3241 if other_props:
3242 util.warn(
3243 "relationship '%s' will copy column %s to column %s, "
3244 "which conflicts with relationship(s): %s. "
3245 "If this is not the intention, consider if these "
3246 "relationships should be linked with "
3247 "back_populates, or if viewonly=True should be "
3248 "applied to one or more if they are read-only. "
3249 "For the less common case that foreign key "
3250 "constraints are partially overlapping, the "
3251 "orm.foreign() "
3252 "annotation can be used to isolate the columns that "
3253 "should be written towards. To silence this "
3254 "warning, add the parameter 'overlaps=\"%s\"' to the "
3255 "'%s' relationship."
3256 % (
3257 self.prop,
3258 from_,
3259 to_,
3260 ", ".join(
3261 sorted(
3262 "'%s' (copies %s to %s)" % (pr, fr_, to_)
3263 for (pr, fr_) in other_props
3264 )
3265 ),
3266 ",".join(sorted(pr.key for pr, fr in other_props)),
3267 self.prop,
3268 ),
3269 code="qzyx",
3270 )
3271 self._track_overlapping_sync_targets[to_][self.prop] = from_
3273 @util.memoized_property
3274 def remote_columns(self) -> Set[ColumnElement[Any]]:
3275 return self._gather_join_annotations("remote")
3277 @util.memoized_property
3278 def local_columns(self) -> Set[ColumnElement[Any]]:
3279 return self._gather_join_annotations("local")
3281 @util.memoized_property
3282 def foreign_key_columns(self) -> Set[ColumnElement[Any]]:
3283 return self._gather_join_annotations("foreign")
3285 def _gather_join_annotations(
3286 self, annotation: str
3287 ) -> Set[ColumnElement[Any]]:
3288 s = set(
3289 self._gather_columns_with_annotation(self.primaryjoin, annotation)
3290 )
3291 if self.secondaryjoin is not None:
3292 s.update(
3293 self._gather_columns_with_annotation(
3294 self.secondaryjoin, annotation
3295 )
3296 )
3297 return {x._deannotate() for x in s}
3299 def _gather_columns_with_annotation(
3300 self, clause: ColumnElement[Any], *annotation: Iterable[str]
3301 ) -> Set[ColumnElement[Any]]:
3302 annotation_set = set(annotation)
3303 return {
3304 cast(ColumnElement[Any], col)
3305 for col in visitors.iterate(clause, {})
3306 if annotation_set.issubset(col._annotations)
3307 }
3309 @util.memoized_property
3310 def _secondary_lineage_set(self) -> FrozenSet[ColumnElement[Any]]:
3311 if self.secondary is not None:
3312 return frozenset(
3313 itertools.chain(*[c.proxy_set for c in self.secondary.c])
3314 )
3315 else:
3316 return util.EMPTY_SET
3318 def join_targets(
3319 self,
3320 source_selectable: Optional[FromClause],
3321 dest_selectable: FromClause,
3322 aliased: bool,
3323 single_crit: Optional[ColumnElement[bool]] = None,
3324 extra_criteria: Tuple[ColumnElement[bool], ...] = (),
3325 ) -> Tuple[
3326 ColumnElement[bool],
3327 Optional[ColumnElement[bool]],
3328 Optional[FromClause],
3329 Optional[ClauseAdapter],
3330 FromClause,
3331 ]:
3332 """Given a source and destination selectable, create a
3333 join between them.
3335 This takes into account aliasing the join clause
3336 to reference the appropriate corresponding columns
3337 in the target objects, as well as the extra child
3338 criterion, equivalent column sets, etc.
3340 """
3341 # place a barrier on the destination such that
3342 # replacement traversals won't ever dig into it.
3343 # its internal structure remains fixed
3344 # regardless of context.
3345 dest_selectable = _shallow_annotate(
3346 dest_selectable, {"no_replacement_traverse": True}
3347 )
3349 primaryjoin, secondaryjoin, secondary = (
3350 self.primaryjoin,
3351 self.secondaryjoin,
3352 self.secondary,
3353 )
3355 # adjust the join condition for single table inheritance,
3356 # in the case that the join is to a subclass
3357 # this is analogous to the
3358 # "_adjust_for_single_table_inheritance()" method in Query.
3360 if single_crit is not None:
3361 if secondaryjoin is not None:
3362 secondaryjoin = secondaryjoin & single_crit
3363 else:
3364 primaryjoin = primaryjoin & single_crit
3366 if extra_criteria:
3368 def mark_exclude_cols(
3369 elem: SupportsAnnotations, annotations: _AnnotationDict
3370 ) -> SupportsAnnotations:
3371 """note unrelated columns in the "extra criteria" as either
3372 should be adapted or not adapted, even though they are not
3373 part of our "local" or "remote" side.
3375 see #9779 for this case, as well as #11010 for a follow up
3377 """
3379 parentmapper_for_element = elem._annotations.get(
3380 "parentmapper", None
3381 )
3383 if (
3384 parentmapper_for_element is not self.prop.parent
3385 and parentmapper_for_element is not self.prop.mapper
3386 and elem not in self._secondary_lineage_set
3387 ):
3388 return _safe_annotate(elem, annotations)
3389 else:
3390 return elem
3392 extra_criteria = tuple(
3393 _deep_annotate(
3394 elem,
3395 {"should_not_adapt": True},
3396 annotate_callable=mark_exclude_cols,
3397 )
3398 for elem in extra_criteria
3399 )
3401 if secondaryjoin is not None:
3402 secondaryjoin = secondaryjoin & sql.and_(*extra_criteria)
3403 else:
3404 primaryjoin = primaryjoin & sql.and_(*extra_criteria)
3406 if aliased:
3407 if secondary is not None:
3408 secondary = secondary._anonymous_fromclause(flat=True)
3409 primary_aliasizer = ClauseAdapter(
3410 secondary,
3411 exclude_fn=_local_col_exclude,
3412 )
3413 secondary_aliasizer = ClauseAdapter(
3414 dest_selectable, equivalents=self.child_equivalents
3415 ).chain(primary_aliasizer)
3416 if source_selectable is not None:
3417 primary_aliasizer = ClauseAdapter(
3418 secondary,
3419 exclude_fn=_local_col_exclude,
3420 ).chain(
3421 ClauseAdapter(
3422 source_selectable,
3423 equivalents=self.parent_equivalents,
3424 )
3425 )
3427 secondaryjoin = secondary_aliasizer.traverse(secondaryjoin)
3428 else:
3429 primary_aliasizer = ClauseAdapter(
3430 dest_selectable,
3431 exclude_fn=_local_col_exclude,
3432 equivalents=self.child_equivalents,
3433 )
3434 if source_selectable is not None:
3435 primary_aliasizer.chain(
3436 ClauseAdapter(
3437 source_selectable,
3438 exclude_fn=_remote_col_exclude,
3439 equivalents=self.parent_equivalents,
3440 )
3441 )
3442 secondary_aliasizer = None
3444 primaryjoin = primary_aliasizer.traverse(primaryjoin)
3445 target_adapter = secondary_aliasizer or primary_aliasizer
3446 target_adapter.exclude_fn = None
3447 else:
3448 target_adapter = None
3449 return (
3450 primaryjoin,
3451 secondaryjoin,
3452 secondary,
3453 target_adapter,
3454 dest_selectable,
3455 )
3457 def create_lazy_clause(self, reverse_direction: bool = False) -> Tuple[
3458 ColumnElement[bool],
3459 Dict[str, ColumnElement[Any]],
3460 Dict[ColumnElement[Any], ColumnElement[Any]],
3461 ]:
3462 binds: Dict[ColumnElement[Any], BindParameter[Any]] = {}
3463 equated_columns: Dict[ColumnElement[Any], ColumnElement[Any]] = {}
3465 has_secondary = self.secondaryjoin is not None
3467 if has_secondary:
3468 lookup = collections.defaultdict(list)
3469 for l, r in self.local_remote_pairs:
3470 lookup[l].append((l, r))
3471 equated_columns[r] = l
3472 elif not reverse_direction:
3473 for l, r in self.local_remote_pairs:
3474 equated_columns[r] = l
3475 else:
3476 for l, r in self.local_remote_pairs:
3477 equated_columns[l] = r
3479 def col_to_bind(
3480 element: ColumnElement[Any], **kw: Any
3481 ) -> Optional[BindParameter[Any]]:
3482 if (
3483 (not reverse_direction and "local" in element._annotations)
3484 or reverse_direction
3485 and (
3486 (has_secondary and element in lookup)
3487 or (not has_secondary and "remote" in element._annotations)
3488 )
3489 ):
3490 if element not in binds:
3491 binds[element] = sql.bindparam(
3492 None, None, type_=element.type, unique=True
3493 )
3494 return binds[element]
3495 return None
3497 lazywhere = self.primaryjoin
3498 if self.secondaryjoin is None or not reverse_direction:
3499 lazywhere = visitors.replacement_traverse(
3500 lazywhere, {}, col_to_bind
3501 )
3503 if self.secondaryjoin is not None:
3504 secondaryjoin = self.secondaryjoin
3505 if reverse_direction:
3506 secondaryjoin = visitors.replacement_traverse(
3507 secondaryjoin, {}, col_to_bind
3508 )
3509 lazywhere = sql.and_(lazywhere, secondaryjoin)
3511 bind_to_col = {binds[col].key: col for col in binds}
3513 return lazywhere, bind_to_col, equated_columns
3516class _ColInAnnotations:
3517 """Serializable object that tests for names in c._annotations.
3519 TODO: does this need to be serializable anymore? can we find what the
3520 use case was for that?
3522 """
3524 __slots__ = ("names",)
3526 def __init__(self, *names: str):
3527 self.names = frozenset(names)
3529 def __call__(self, c: ClauseElement) -> bool:
3530 return bool(self.names.intersection(c._annotations))
3533_local_col_exclude = _ColInAnnotations("local", "should_not_adapt")
3534_remote_col_exclude = _ColInAnnotations("remote", "should_not_adapt")
3537class Relationship(
3538 RelationshipProperty[_T],
3539 _DeclarativeMapped[_T],
3540):
3541 """Describes an object property that holds a single item or list
3542 of items that correspond to a related database table.
3544 Public constructor is the :func:`_orm.relationship` function.
3546 .. seealso::
3548 :ref:`relationship_config_toplevel`
3550 .. versionchanged:: 2.0 Added :class:`_orm.Relationship` as a Declarative
3551 compatible subclass for :class:`_orm.RelationshipProperty`.
3553 """
3555 inherit_cache = True
3556 """:meta private:"""
3559class _RelationshipDeclared( # type: ignore[misc]
3560 Relationship[_T],
3561 WriteOnlyMapped[_T], # not compatible with Mapped[_T]
3562 DynamicMapped[_T], # not compatible with Mapped[_T]
3563):
3564 """Relationship subclass used implicitly for declarative mapping."""
3566 inherit_cache = True
3567 """:meta private:"""
3569 @classmethod
3570 def _mapper_property_name(cls) -> str:
3571 return "Relationship"