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