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