diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/orm/relationships.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/sqlalchemy/orm/relationships.py | 3500 |
1 files changed, 0 insertions, 3500 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/orm/relationships.py b/venv/lib/python3.11/site-packages/sqlalchemy/orm/relationships.py deleted file mode 100644 index b5e33ff..0000000 --- a/venv/lib/python3.11/site-packages/sqlalchemy/orm/relationships.py +++ /dev/null @@ -1,3500 +0,0 @@ -# orm/relationships.py -# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors -# <see AUTHORS file> -# -# This module is part of SQLAlchemy and is released under -# the MIT License: https://www.opensource.org/licenses/mit-license.php - -"""Heuristics related to join conditions as used in -:func:`_orm.relationship`. - -Provides the :class:`.JoinCondition` object, which encapsulates -SQL annotation and aliasing behavior focused on the `primaryjoin` -and `secondaryjoin` aspects of :func:`_orm.relationship`. - -""" -from __future__ import annotations - -import collections -from collections import abc -import dataclasses -import inspect as _py_inspect -import itertools -import re -import typing -from typing import Any -from typing import Callable -from typing import cast -from typing import Collection -from typing import Dict -from typing import FrozenSet -from typing import Generic -from typing import Iterable -from typing import Iterator -from typing import List -from typing import NamedTuple -from typing import NoReturn -from typing import Optional -from typing import Sequence -from typing import Set -from typing import Tuple -from typing import Type -from typing import TypeVar -from typing import Union -import weakref - -from . import attributes -from . import strategy_options -from ._typing import insp_is_aliased_class -from ._typing import is_has_collection_adapter -from .base import _DeclarativeMapped -from .base import _is_mapped_class -from .base import class_mapper -from .base import DynamicMapped -from .base import LoaderCallableStatus -from .base import PassiveFlag -from .base import state_str -from .base import WriteOnlyMapped -from .interfaces import _AttributeOptions -from .interfaces import _IntrospectsAnnotations -from .interfaces import MANYTOMANY -from .interfaces import MANYTOONE -from .interfaces import ONETOMANY -from .interfaces import PropComparator -from .interfaces import RelationshipDirection -from .interfaces import StrategizedProperty -from .util import _orm_annotate -from .util import _orm_deannotate -from .util import CascadeOptions -from .. import exc as sa_exc -from .. import Exists -from .. import log -from .. import schema -from .. import sql -from .. import util -from ..inspection import inspect -from ..sql import coercions -from ..sql import expression -from ..sql import operators -from ..sql import roles -from ..sql import visitors -from ..sql._typing import _ColumnExpressionArgument -from ..sql._typing import _HasClauseElement -from ..sql.annotation import _safe_annotate -from ..sql.elements import ColumnClause -from ..sql.elements import ColumnElement -from ..sql.util import _deep_annotate -from ..sql.util import _deep_deannotate -from ..sql.util import _shallow_annotate -from ..sql.util import adapt_criterion_to_null -from ..sql.util import ClauseAdapter -from ..sql.util import join_condition -from ..sql.util import selectables_overlap -from ..sql.util import visit_binary_product -from ..util.typing import de_optionalize_union_types -from ..util.typing import Literal -from ..util.typing import resolve_name_to_real_class_name - -if typing.TYPE_CHECKING: - from ._typing import _EntityType - from ._typing import _ExternalEntityType - from ._typing import _IdentityKeyType - from ._typing import _InstanceDict - from ._typing import _InternalEntityType - from ._typing import _O - from ._typing import _RegistryType - from .base import Mapped - from .clsregistry import _class_resolver - from .clsregistry import _ModNS - from .decl_base import _ClassScanMapperConfig - from .dependency import DependencyProcessor - from .mapper import Mapper - from .query import Query - from .session import Session - from .state import InstanceState - from .strategies import LazyLoader - from .util import AliasedClass - from .util import AliasedInsp - from ..sql._typing import _CoreAdapterProto - from ..sql._typing import _EquivalentColumnMap - from ..sql._typing import _InfoType - from ..sql.annotation import _AnnotationDict - from ..sql.annotation import SupportsAnnotations - from ..sql.elements import BinaryExpression - from ..sql.elements import BindParameter - from ..sql.elements import ClauseElement - from ..sql.schema import Table - from ..sql.selectable import FromClause - from ..util.typing import _AnnotationScanType - from ..util.typing import RODescriptorReference - -_T = TypeVar("_T", bound=Any) -_T1 = TypeVar("_T1", bound=Any) -_T2 = TypeVar("_T2", bound=Any) - -_PT = TypeVar("_PT", bound=Any) - -_PT2 = TypeVar("_PT2", bound=Any) - - -_RelationshipArgumentType = Union[ - str, - Type[_T], - Callable[[], Type[_T]], - "Mapper[_T]", - "AliasedClass[_T]", - Callable[[], "Mapper[_T]"], - Callable[[], "AliasedClass[_T]"], -] - -_LazyLoadArgumentType = Literal[ - "select", - "joined", - "selectin", - "subquery", - "raise", - "raise_on_sql", - "noload", - "immediate", - "write_only", - "dynamic", - True, - False, - None, -] - - -_RelationshipJoinConditionArgument = Union[ - str, _ColumnExpressionArgument[bool] -] -_RelationshipSecondaryArgument = Union[ - "FromClause", str, Callable[[], "FromClause"] -] -_ORMOrderByArgument = Union[ - Literal[False], - str, - _ColumnExpressionArgument[Any], - Callable[[], _ColumnExpressionArgument[Any]], - Callable[[], Iterable[_ColumnExpressionArgument[Any]]], - Iterable[Union[str, _ColumnExpressionArgument[Any]]], -] -ORMBackrefArgument = Union[str, Tuple[str, Dict[str, Any]]] - -_ORMColCollectionElement = Union[ - ColumnClause[Any], - _HasClauseElement[Any], - roles.DMLColumnRole, - "Mapped[Any]", -] -_ORMColCollectionArgument = Union[ - str, - Sequence[_ORMColCollectionElement], - Callable[[], Sequence[_ORMColCollectionElement]], - Callable[[], _ORMColCollectionElement], - _ORMColCollectionElement, -] - - -_CEA = TypeVar("_CEA", bound=_ColumnExpressionArgument[Any]) - -_CE = TypeVar("_CE", bound="ColumnElement[Any]") - - -_ColumnPairIterable = Iterable[Tuple[ColumnElement[Any], ColumnElement[Any]]] - -_ColumnPairs = Sequence[Tuple[ColumnElement[Any], ColumnElement[Any]]] - -_MutableColumnPairs = List[Tuple[ColumnElement[Any], ColumnElement[Any]]] - - -def remote(expr: _CEA) -> _CEA: - """Annotate a portion of a primaryjoin expression - with a 'remote' annotation. - - See the section :ref:`relationship_custom_foreign` for a - description of use. - - .. seealso:: - - :ref:`relationship_custom_foreign` - - :func:`.foreign` - - """ - return _annotate_columns( # type: ignore - coercions.expect(roles.ColumnArgumentRole, expr), {"remote": True} - ) - - -def foreign(expr: _CEA) -> _CEA: - """Annotate a portion of a primaryjoin expression - with a 'foreign' annotation. - - See the section :ref:`relationship_custom_foreign` for a - description of use. - - .. seealso:: - - :ref:`relationship_custom_foreign` - - :func:`.remote` - - """ - - return _annotate_columns( # type: ignore - coercions.expect(roles.ColumnArgumentRole, expr), {"foreign": True} - ) - - -@dataclasses.dataclass -class _RelationshipArg(Generic[_T1, _T2]): - """stores a user-defined parameter value that must be resolved and - parsed later at mapper configuration time. - - """ - - __slots__ = "name", "argument", "resolved" - name: str - argument: _T1 - resolved: Optional[_T2] - - def _is_populated(self) -> bool: - return self.argument is not None - - def _resolve_against_registry( - self, clsregistry_resolver: Callable[[str, bool], _class_resolver] - ) -> None: - attr_value = self.argument - - if isinstance(attr_value, str): - self.resolved = clsregistry_resolver( - attr_value, self.name == "secondary" - )() - elif callable(attr_value) and not _is_mapped_class(attr_value): - self.resolved = attr_value() - else: - self.resolved = attr_value - - -_RelationshipOrderByArg = Union[Literal[False], Tuple[ColumnElement[Any], ...]] - - -class _RelationshipArgs(NamedTuple): - """stores user-passed parameters that are resolved at mapper configuration - time. - - """ - - secondary: _RelationshipArg[ - Optional[_RelationshipSecondaryArgument], - Optional[FromClause], - ] - primaryjoin: _RelationshipArg[ - Optional[_RelationshipJoinConditionArgument], - Optional[ColumnElement[Any]], - ] - secondaryjoin: _RelationshipArg[ - Optional[_RelationshipJoinConditionArgument], - Optional[ColumnElement[Any]], - ] - order_by: _RelationshipArg[_ORMOrderByArgument, _RelationshipOrderByArg] - foreign_keys: _RelationshipArg[ - Optional[_ORMColCollectionArgument], Set[ColumnElement[Any]] - ] - remote_side: _RelationshipArg[ - Optional[_ORMColCollectionArgument], Set[ColumnElement[Any]] - ] - - -@log.class_logger -class RelationshipProperty( - _IntrospectsAnnotations, StrategizedProperty[_T], log.Identified -): - """Describes an object property that holds a single item or list - of items that correspond to a related database table. - - Public constructor is the :func:`_orm.relationship` function. - - .. seealso:: - - :ref:`relationship_config_toplevel` - - """ - - strategy_wildcard_key = strategy_options._RELATIONSHIP_TOKEN - inherit_cache = True - """:meta private:""" - - _links_to_entity = True - _is_relationship = True - - _overlaps: Sequence[str] - - _lazy_strategy: LazyLoader - - _persistence_only = dict( - passive_deletes=False, - passive_updates=True, - enable_typechecks=True, - active_history=False, - cascade_backrefs=False, - ) - - _dependency_processor: Optional[DependencyProcessor] = None - - primaryjoin: ColumnElement[bool] - secondaryjoin: Optional[ColumnElement[bool]] - secondary: Optional[FromClause] - _join_condition: JoinCondition - order_by: _RelationshipOrderByArg - - _user_defined_foreign_keys: Set[ColumnElement[Any]] - _calculated_foreign_keys: Set[ColumnElement[Any]] - - remote_side: Set[ColumnElement[Any]] - local_columns: Set[ColumnElement[Any]] - - synchronize_pairs: _ColumnPairs - secondary_synchronize_pairs: Optional[_ColumnPairs] - - local_remote_pairs: Optional[_ColumnPairs] - - direction: RelationshipDirection - - _init_args: _RelationshipArgs - - def __init__( - self, - argument: Optional[_RelationshipArgumentType[_T]] = None, - secondary: Optional[_RelationshipSecondaryArgument] = None, - *, - uselist: Optional[bool] = None, - collection_class: Optional[ - Union[Type[Collection[Any]], Callable[[], Collection[Any]]] - ] = None, - primaryjoin: Optional[_RelationshipJoinConditionArgument] = None, - secondaryjoin: Optional[_RelationshipJoinConditionArgument] = None, - back_populates: Optional[str] = None, - order_by: _ORMOrderByArgument = False, - backref: Optional[ORMBackrefArgument] = None, - overlaps: Optional[str] = None, - post_update: bool = False, - cascade: str = "save-update, merge", - viewonly: bool = False, - attribute_options: Optional[_AttributeOptions] = None, - lazy: _LazyLoadArgumentType = "select", - passive_deletes: Union[Literal["all"], bool] = False, - passive_updates: bool = True, - active_history: bool = False, - enable_typechecks: bool = True, - foreign_keys: Optional[_ORMColCollectionArgument] = None, - remote_side: Optional[_ORMColCollectionArgument] = None, - join_depth: Optional[int] = None, - comparator_factory: Optional[ - Type[RelationshipProperty.Comparator[Any]] - ] = None, - single_parent: bool = False, - innerjoin: bool = False, - distinct_target_key: Optional[bool] = None, - load_on_pending: bool = False, - query_class: Optional[Type[Query[Any]]] = None, - info: Optional[_InfoType] = None, - omit_join: Literal[None, False] = None, - sync_backref: Optional[bool] = None, - doc: Optional[str] = None, - bake_queries: Literal[True] = True, - cascade_backrefs: Literal[False] = False, - _local_remote_pairs: Optional[_ColumnPairs] = None, - _legacy_inactive_history_style: bool = False, - ): - super().__init__(attribute_options=attribute_options) - - self.uselist = uselist - self.argument = argument - - self._init_args = _RelationshipArgs( - _RelationshipArg("secondary", secondary, None), - _RelationshipArg("primaryjoin", primaryjoin, None), - _RelationshipArg("secondaryjoin", secondaryjoin, None), - _RelationshipArg("order_by", order_by, None), - _RelationshipArg("foreign_keys", foreign_keys, None), - _RelationshipArg("remote_side", remote_side, None), - ) - - self.post_update = post_update - self.viewonly = viewonly - if viewonly: - self._warn_for_persistence_only_flags( - passive_deletes=passive_deletes, - passive_updates=passive_updates, - enable_typechecks=enable_typechecks, - active_history=active_history, - cascade_backrefs=cascade_backrefs, - ) - if viewonly and sync_backref: - raise sa_exc.ArgumentError( - "sync_backref and viewonly cannot both be True" - ) - self.sync_backref = sync_backref - self.lazy = lazy - self.single_parent = single_parent - self.collection_class = collection_class - self.passive_deletes = passive_deletes - - if cascade_backrefs: - raise sa_exc.ArgumentError( - "The 'cascade_backrefs' parameter passed to " - "relationship() may only be set to False." - ) - - self.passive_updates = passive_updates - self.enable_typechecks = enable_typechecks - self.query_class = query_class - self.innerjoin = innerjoin - self.distinct_target_key = distinct_target_key - self.doc = doc - self.active_history = active_history - self._legacy_inactive_history_style = _legacy_inactive_history_style - - self.join_depth = join_depth - if omit_join: - util.warn( - "setting omit_join to True is not supported; selectin " - "loading of this relationship may not work correctly if this " - "flag is set explicitly. omit_join optimization is " - "automatically detected for conditions under which it is " - "supported." - ) - - self.omit_join = omit_join - self.local_remote_pairs = _local_remote_pairs - self.load_on_pending = load_on_pending - self.comparator_factory = ( - comparator_factory or RelationshipProperty.Comparator - ) - util.set_creation_order(self) - - if info is not None: - self.info.update(info) - - self.strategy_key = (("lazy", self.lazy),) - - self._reverse_property: Set[RelationshipProperty[Any]] = set() - - if overlaps: - self._overlaps = set(re.split(r"\s*,\s*", overlaps)) # type: ignore # noqa: E501 - else: - self._overlaps = () - - # mypy ignoring the @property setter - self.cascade = cascade # type: ignore - - self.back_populates = back_populates - - if self.back_populates: - if backref: - raise sa_exc.ArgumentError( - "backref and back_populates keyword arguments " - "are mutually exclusive" - ) - self.backref = None - else: - self.backref = backref - - def _warn_for_persistence_only_flags(self, **kw: Any) -> None: - for k, v in kw.items(): - if v != self._persistence_only[k]: - # we are warning here rather than warn deprecated as this is a - # configuration mistake, and Python shows regular warnings more - # aggressively than deprecation warnings by default. Unlike the - # case of setting viewonly with cascade, the settings being - # warned about here are not actively doing the wrong thing - # against viewonly=True, so it is not as urgent to have these - # raise an error. - util.warn( - "Setting %s on relationship() while also " - "setting viewonly=True does not make sense, as a " - "viewonly=True relationship does not perform persistence " - "operations. This configuration may raise an error " - "in a future release." % (k,) - ) - - def instrument_class(self, mapper: Mapper[Any]) -> None: - attributes.register_descriptor( - mapper.class_, - self.key, - comparator=self.comparator_factory(self, mapper), - parententity=mapper, - doc=self.doc, - ) - - class Comparator(util.MemoizedSlots, PropComparator[_PT]): - """Produce boolean, comparison, and other operators for - :class:`.RelationshipProperty` attributes. - - See the documentation for :class:`.PropComparator` for a brief - overview of ORM level operator definition. - - .. seealso:: - - :class:`.PropComparator` - - :class:`.ColumnProperty.Comparator` - - :class:`.ColumnOperators` - - :ref:`types_operators` - - :attr:`.TypeEngine.comparator_factory` - - """ - - __slots__ = ( - "entity", - "mapper", - "property", - "_of_type", - "_extra_criteria", - ) - - prop: RODescriptorReference[RelationshipProperty[_PT]] - _of_type: Optional[_EntityType[_PT]] - - def __init__( - self, - prop: RelationshipProperty[_PT], - parentmapper: _InternalEntityType[Any], - adapt_to_entity: Optional[AliasedInsp[Any]] = None, - of_type: Optional[_EntityType[_PT]] = None, - extra_criteria: Tuple[ColumnElement[bool], ...] = (), - ): - """Construction of :class:`.RelationshipProperty.Comparator` - is internal to the ORM's attribute mechanics. - - """ - self.prop = prop - self._parententity = parentmapper - self._adapt_to_entity = adapt_to_entity - if of_type: - self._of_type = of_type - else: - self._of_type = None - self._extra_criteria = extra_criteria - - def adapt_to_entity( - self, adapt_to_entity: AliasedInsp[Any] - ) -> RelationshipProperty.Comparator[Any]: - return self.__class__( - self.prop, - self._parententity, - adapt_to_entity=adapt_to_entity, - of_type=self._of_type, - ) - - entity: _InternalEntityType[_PT] - """The target entity referred to by this - :class:`.RelationshipProperty.Comparator`. - - This is either a :class:`_orm.Mapper` or :class:`.AliasedInsp` - object. - - This is the "target" or "remote" side of the - :func:`_orm.relationship`. - - """ - - mapper: Mapper[_PT] - """The target :class:`_orm.Mapper` referred to by this - :class:`.RelationshipProperty.Comparator`. - - This is the "target" or "remote" side of the - :func:`_orm.relationship`. - - """ - - def _memoized_attr_entity(self) -> _InternalEntityType[_PT]: - if self._of_type: - return inspect(self._of_type) # type: ignore - else: - return self.prop.entity - - def _memoized_attr_mapper(self) -> Mapper[_PT]: - return self.entity.mapper - - def _source_selectable(self) -> FromClause: - if self._adapt_to_entity: - return self._adapt_to_entity.selectable - else: - return self.property.parent._with_polymorphic_selectable - - def __clause_element__(self) -> ColumnElement[bool]: - adapt_from = self._source_selectable() - if self._of_type: - of_type_entity = inspect(self._of_type) - else: - of_type_entity = None - - ( - pj, - sj, - source, - dest, - secondary, - target_adapter, - ) = self.prop._create_joins( - source_selectable=adapt_from, - source_polymorphic=True, - of_type_entity=of_type_entity, - alias_secondary=True, - extra_criteria=self._extra_criteria, - ) - if sj is not None: - return pj & sj - else: - return pj - - def of_type(self, class_: _EntityType[Any]) -> PropComparator[_PT]: - r"""Redefine this object in terms of a polymorphic subclass. - - See :meth:`.PropComparator.of_type` for an example. - - - """ - return RelationshipProperty.Comparator( - self.prop, - self._parententity, - adapt_to_entity=self._adapt_to_entity, - of_type=class_, - extra_criteria=self._extra_criteria, - ) - - def and_( - self, *criteria: _ColumnExpressionArgument[bool] - ) -> PropComparator[Any]: - """Add AND criteria. - - See :meth:`.PropComparator.and_` for an example. - - .. versionadded:: 1.4 - - """ - exprs = tuple( - coercions.expect(roles.WhereHavingRole, clause) - for clause in util.coerce_generator_arg(criteria) - ) - - return RelationshipProperty.Comparator( - self.prop, - self._parententity, - adapt_to_entity=self._adapt_to_entity, - of_type=self._of_type, - extra_criteria=self._extra_criteria + exprs, - ) - - def in_(self, other: Any) -> NoReturn: - """Produce an IN clause - this is not implemented - for :func:`_orm.relationship`-based attributes at this time. - - """ - raise NotImplementedError( - "in_() not yet supported for " - "relationships. For a simple " - "many-to-one, use in_() against " - "the set of foreign key values." - ) - - # https://github.com/python/mypy/issues/4266 - __hash__ = None # type: ignore - - def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501 - """Implement the ``==`` operator. - - In a many-to-one context, such as:: - - MyClass.some_prop == <some object> - - this will typically produce a - clause such as:: - - mytable.related_id == <some id> - - Where ``<some id>`` is the primary key of the given - object. - - The ``==`` operator provides partial functionality for non- - many-to-one comparisons: - - * Comparisons against collections are not supported. - Use :meth:`~.Relationship.Comparator.contains`. - * Compared to a scalar one-to-many, will produce a - clause that compares the target columns in the parent to - the given target. - * Compared to a scalar many-to-many, an alias - of the association table will be rendered as - well, forming a natural join that is part of the - main body of the query. This will not work for - queries that go beyond simple AND conjunctions of - comparisons, such as those which use OR. Use - explicit joins, outerjoins, or - :meth:`~.Relationship.Comparator.has` for - more comprehensive non-many-to-one scalar - membership tests. - * Comparisons against ``None`` given in a one-to-many - or many-to-many context produce a NOT EXISTS clause. - - """ - if other is None or isinstance(other, expression.Null): - if self.property.direction in [ONETOMANY, MANYTOMANY]: - return ~self._criterion_exists() - else: - return _orm_annotate( - self.property._optimized_compare( - None, adapt_source=self.adapter - ) - ) - elif self.property.uselist: - raise sa_exc.InvalidRequestError( - "Can't compare a collection to an object or collection; " - "use contains() to test for membership." - ) - else: - return _orm_annotate( - self.property._optimized_compare( - other, adapt_source=self.adapter - ) - ) - - def _criterion_exists( - self, - criterion: Optional[_ColumnExpressionArgument[bool]] = None, - **kwargs: Any, - ) -> Exists: - where_criteria = ( - coercions.expect(roles.WhereHavingRole, criterion) - if criterion is not None - else None - ) - - if getattr(self, "_of_type", None): - info: Optional[_InternalEntityType[Any]] = inspect( - self._of_type - ) - assert info is not None - target_mapper, to_selectable, is_aliased_class = ( - info.mapper, - info.selectable, - info.is_aliased_class, - ) - if self.property._is_self_referential and not is_aliased_class: - to_selectable = to_selectable._anonymous_fromclause() - - single_crit = target_mapper._single_table_criterion - if single_crit is not None: - if where_criteria is not None: - where_criteria = single_crit & where_criteria - else: - where_criteria = single_crit - else: - is_aliased_class = False - to_selectable = None - - if self.adapter: - source_selectable = self._source_selectable() - else: - source_selectable = None - - ( - pj, - sj, - source, - dest, - secondary, - target_adapter, - ) = self.property._create_joins( - dest_selectable=to_selectable, - source_selectable=source_selectable, - ) - - for k in kwargs: - crit = getattr(self.property.mapper.class_, k) == kwargs[k] - if where_criteria is None: - where_criteria = crit - else: - where_criteria = where_criteria & crit - - # annotate the *local* side of the join condition, in the case - # of pj + sj this is the full primaryjoin, in the case of just - # pj its the local side of the primaryjoin. - if sj is not None: - j = _orm_annotate(pj) & sj - else: - j = _orm_annotate(pj, exclude=self.property.remote_side) - - if ( - where_criteria is not None - and target_adapter - and not is_aliased_class - ): - # limit this adapter to annotated only? - where_criteria = target_adapter.traverse(where_criteria) - - # only have the "joined left side" of what we - # return be subject to Query adaption. The right - # side of it is used for an exists() subquery and - # should not correlate or otherwise reach out - # to anything in the enclosing query. - if where_criteria is not None: - where_criteria = where_criteria._annotate( - {"no_replacement_traverse": True} - ) - - crit = j & sql.True_._ifnone(where_criteria) - - if secondary is not None: - ex = ( - sql.exists(1) - .where(crit) - .select_from(dest, secondary) - .correlate_except(dest, secondary) - ) - else: - ex = ( - sql.exists(1) - .where(crit) - .select_from(dest) - .correlate_except(dest) - ) - return ex - - def any( - self, - criterion: Optional[_ColumnExpressionArgument[bool]] = None, - **kwargs: Any, - ) -> ColumnElement[bool]: - """Produce an expression that tests a collection against - particular criterion, using EXISTS. - - An expression like:: - - session.query(MyClass).filter( - MyClass.somereference.any(SomeRelated.x==2) - ) - - - Will produce a query like:: - - SELECT * FROM my_table WHERE - EXISTS (SELECT 1 FROM related WHERE related.my_id=my_table.id - AND related.x=2) - - Because :meth:`~.Relationship.Comparator.any` uses - a correlated subquery, its performance is not nearly as - good when compared against large target tables as that of - using a join. - - :meth:`~.Relationship.Comparator.any` is particularly - useful for testing for empty collections:: - - session.query(MyClass).filter( - ~MyClass.somereference.any() - ) - - will produce:: - - SELECT * FROM my_table WHERE - NOT (EXISTS (SELECT 1 FROM related WHERE - related.my_id=my_table.id)) - - :meth:`~.Relationship.Comparator.any` is only - valid for collections, i.e. a :func:`_orm.relationship` - that has ``uselist=True``. For scalar references, - use :meth:`~.Relationship.Comparator.has`. - - """ - if not self.property.uselist: - raise sa_exc.InvalidRequestError( - "'any()' not implemented for scalar " - "attributes. Use has()." - ) - - return self._criterion_exists(criterion, **kwargs) - - def has( - self, - criterion: Optional[_ColumnExpressionArgument[bool]] = None, - **kwargs: Any, - ) -> ColumnElement[bool]: - """Produce an expression that tests a scalar reference against - particular criterion, using EXISTS. - - An expression like:: - - session.query(MyClass).filter( - MyClass.somereference.has(SomeRelated.x==2) - ) - - - Will produce a query like:: - - SELECT * FROM my_table WHERE - EXISTS (SELECT 1 FROM related WHERE - related.id==my_table.related_id AND related.x=2) - - Because :meth:`~.Relationship.Comparator.has` uses - a correlated subquery, its performance is not nearly as - good when compared against large target tables as that of - using a join. - - :meth:`~.Relationship.Comparator.has` is only - valid for scalar references, i.e. a :func:`_orm.relationship` - that has ``uselist=False``. For collection references, - use :meth:`~.Relationship.Comparator.any`. - - """ - if self.property.uselist: - raise sa_exc.InvalidRequestError( - "'has()' not implemented for collections. Use any()." - ) - return self._criterion_exists(criterion, **kwargs) - - def contains( - self, other: _ColumnExpressionArgument[Any], **kwargs: Any - ) -> ColumnElement[bool]: - """Return a simple expression that tests a collection for - containment of a particular item. - - :meth:`~.Relationship.Comparator.contains` is - only valid for a collection, i.e. a - :func:`_orm.relationship` that implements - one-to-many or many-to-many with ``uselist=True``. - - When used in a simple one-to-many context, an - expression like:: - - MyClass.contains(other) - - Produces a clause like:: - - mytable.id == <some id> - - Where ``<some id>`` is the value of the foreign key - attribute on ``other`` which refers to the primary - key of its parent object. From this it follows that - :meth:`~.Relationship.Comparator.contains` is - very useful when used with simple one-to-many - operations. - - For many-to-many operations, the behavior of - :meth:`~.Relationship.Comparator.contains` - has more caveats. The association table will be - rendered in the statement, producing an "implicit" - join, that is, includes multiple tables in the FROM - clause which are equated in the WHERE clause:: - - query(MyClass).filter(MyClass.contains(other)) - - Produces a query like:: - - SELECT * FROM my_table, my_association_table AS - my_association_table_1 WHERE - my_table.id = my_association_table_1.parent_id - AND my_association_table_1.child_id = <some id> - - Where ``<some id>`` would be the primary key of - ``other``. From the above, it is clear that - :meth:`~.Relationship.Comparator.contains` - will **not** work with many-to-many collections when - used in queries that move beyond simple AND - conjunctions, such as multiple - :meth:`~.Relationship.Comparator.contains` - expressions joined by OR. In such cases subqueries or - explicit "outer joins" will need to be used instead. - See :meth:`~.Relationship.Comparator.any` for - a less-performant alternative using EXISTS, or refer - to :meth:`_query.Query.outerjoin` - as well as :ref:`orm_queryguide_joins` - for more details on constructing outer joins. - - kwargs may be ignored by this operator but are required for API - conformance. - """ - if not self.prop.uselist: - raise sa_exc.InvalidRequestError( - "'contains' not implemented for scalar " - "attributes. Use ==" - ) - - clause = self.prop._optimized_compare( - other, adapt_source=self.adapter - ) - - if self.prop.secondaryjoin is not None: - clause.negation_clause = self.__negated_contains_or_equals( - other - ) - - return clause - - def __negated_contains_or_equals( - self, other: Any - ) -> ColumnElement[bool]: - if self.prop.direction == MANYTOONE: - state = attributes.instance_state(other) - - def state_bindparam( - local_col: ColumnElement[Any], - state: InstanceState[Any], - remote_col: ColumnElement[Any], - ) -> BindParameter[Any]: - dict_ = state.dict - return sql.bindparam( - local_col.key, - type_=local_col.type, - unique=True, - callable_=self.prop._get_attr_w_warn_on_none( - self.prop.mapper, state, dict_, remote_col - ), - ) - - def adapt(col: _CE) -> _CE: - if self.adapter: - return self.adapter(col) - else: - return col - - if self.property._use_get: - return sql.and_( - *[ - sql.or_( - adapt(x) - != state_bindparam(adapt(x), state, y), - adapt(x) == None, - ) - for (x, y) in self.property.local_remote_pairs - ] - ) - - criterion = sql.and_( - *[ - x == y - for (x, y) in zip( - self.property.mapper.primary_key, - self.property.mapper.primary_key_from_instance(other), - ) - ] - ) - - return ~self._criterion_exists(criterion) - - def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501 - """Implement the ``!=`` operator. - - In a many-to-one context, such as:: - - MyClass.some_prop != <some object> - - This will typically produce a clause such as:: - - mytable.related_id != <some id> - - Where ``<some id>`` is the primary key of the - given object. - - The ``!=`` operator provides partial functionality for non- - many-to-one comparisons: - - * Comparisons against collections are not supported. - Use - :meth:`~.Relationship.Comparator.contains` - in conjunction with :func:`_expression.not_`. - * Compared to a scalar one-to-many, will produce a - clause that compares the target columns in the parent to - the given target. - * Compared to a scalar many-to-many, an alias - of the association table will be rendered as - well, forming a natural join that is part of the - main body of the query. This will not work for - queries that go beyond simple AND conjunctions of - comparisons, such as those which use OR. Use - explicit joins, outerjoins, or - :meth:`~.Relationship.Comparator.has` in - conjunction with :func:`_expression.not_` for - more comprehensive non-many-to-one scalar - membership tests. - * Comparisons against ``None`` given in a one-to-many - or many-to-many context produce an EXISTS clause. - - """ - if other is None or isinstance(other, expression.Null): - if self.property.direction == MANYTOONE: - return _orm_annotate( - ~self.property._optimized_compare( - None, adapt_source=self.adapter - ) - ) - - else: - return self._criterion_exists() - elif self.property.uselist: - raise sa_exc.InvalidRequestError( - "Can't compare a collection" - " to an object or collection; use " - "contains() to test for membership." - ) - else: - return _orm_annotate(self.__negated_contains_or_equals(other)) - - def _memoized_attr_property(self) -> RelationshipProperty[_PT]: - self.prop.parent._check_configure() - return self.prop - - def _with_parent( - self, - instance: object, - alias_secondary: bool = True, - from_entity: Optional[_EntityType[Any]] = None, - ) -> ColumnElement[bool]: - assert instance is not None - adapt_source: Optional[_CoreAdapterProto] = None - if from_entity is not None: - insp: Optional[_InternalEntityType[Any]] = inspect(from_entity) - assert insp is not None - if insp_is_aliased_class(insp): - adapt_source = insp._adapter.adapt_clause - return self._optimized_compare( - instance, - value_is_parent=True, - adapt_source=adapt_source, - alias_secondary=alias_secondary, - ) - - def _optimized_compare( - self, - state: Any, - value_is_parent: bool = False, - adapt_source: Optional[_CoreAdapterProto] = None, - alias_secondary: bool = True, - ) -> ColumnElement[bool]: - if state is not None: - try: - state = inspect(state) - except sa_exc.NoInspectionAvailable: - state = None - - if state is None or not getattr(state, "is_instance", False): - raise sa_exc.ArgumentError( - "Mapped instance expected for relationship " - "comparison to object. Classes, queries and other " - "SQL elements are not accepted in this context; for " - "comparison with a subquery, " - "use %s.has(**criteria)." % self - ) - reverse_direction = not value_is_parent - - if state is None: - return self._lazy_none_clause( - reverse_direction, adapt_source=adapt_source - ) - - if not reverse_direction: - criterion, bind_to_col = ( - self._lazy_strategy._lazywhere, - self._lazy_strategy._bind_to_col, - ) - else: - criterion, bind_to_col = ( - self._lazy_strategy._rev_lazywhere, - self._lazy_strategy._rev_bind_to_col, - ) - - if reverse_direction: - mapper = self.mapper - else: - mapper = self.parent - - dict_ = attributes.instance_dict(state.obj()) - - def visit_bindparam(bindparam: BindParameter[Any]) -> None: - if bindparam._identifying_key in bind_to_col: - bindparam.callable = self._get_attr_w_warn_on_none( - mapper, - state, - dict_, - bind_to_col[bindparam._identifying_key], - ) - - if self.secondary is not None and alias_secondary: - criterion = ClauseAdapter( - self.secondary._anonymous_fromclause() - ).traverse(criterion) - - criterion = visitors.cloned_traverse( - criterion, {}, {"bindparam": visit_bindparam} - ) - - if adapt_source: - criterion = adapt_source(criterion) - return criterion - - def _get_attr_w_warn_on_none( - self, - mapper: Mapper[Any], - state: InstanceState[Any], - dict_: _InstanceDict, - column: ColumnElement[Any], - ) -> Callable[[], Any]: - """Create the callable that is used in a many-to-one expression. - - E.g.:: - - u1 = s.query(User).get(5) - - expr = Address.user == u1 - - Above, the SQL should be "address.user_id = 5". The callable - returned by this method produces the value "5" based on the identity - of ``u1``. - - """ - - # in this callable, we're trying to thread the needle through - # a wide variety of scenarios, including: - # - # * the object hasn't been flushed yet and there's no value for - # the attribute as of yet - # - # * the object hasn't been flushed yet but it has a user-defined - # value - # - # * the object has a value but it's expired and not locally present - # - # * the object has a value but it's expired and not locally present, - # and the object is also detached - # - # * The object hadn't been flushed yet, there was no value, but - # later, the object has been expired and detached, and *now* - # they're trying to evaluate it - # - # * the object had a value, but it was changed to a new value, and - # then expired - # - # * the object had a value, but it was changed to a new value, and - # then expired, then the object was detached - # - # * the object has a user-set value, but it's None and we don't do - # the comparison correctly for that so warn - # - - prop = mapper.get_property_by_column(column) - - # by invoking this method, InstanceState will track the last known - # value for this key each time the attribute is to be expired. - # this feature was added explicitly for use in this method. - state._track_last_known_value(prop.key) - - lkv_fixed = state._last_known_values - - def _go() -> Any: - assert lkv_fixed is not None - last_known = to_return = lkv_fixed[prop.key] - existing_is_available = ( - last_known is not LoaderCallableStatus.NO_VALUE - ) - - # we support that the value may have changed. so here we - # try to get the most recent value including re-fetching. - # only if we can't get a value now due to detachment do we return - # the last known value - current_value = mapper._get_state_attr_by_column( - state, - dict_, - column, - passive=( - PassiveFlag.PASSIVE_OFF - if state.persistent - else PassiveFlag.PASSIVE_NO_FETCH ^ PassiveFlag.INIT_OK - ), - ) - - if current_value is LoaderCallableStatus.NEVER_SET: - if not existing_is_available: - raise sa_exc.InvalidRequestError( - "Can't resolve value for column %s on object " - "%s; no value has been set for this column" - % (column, state_str(state)) - ) - elif current_value is LoaderCallableStatus.PASSIVE_NO_RESULT: - if not existing_is_available: - raise sa_exc.InvalidRequestError( - "Can't resolve value for column %s on object " - "%s; the object is detached and the value was " - "expired" % (column, state_str(state)) - ) - else: - to_return = current_value - if to_return is None: - util.warn( - "Got None for value of column %s; this is unsupported " - "for a relationship comparison and will not " - "currently produce an IS comparison " - "(but may in a future release)" % column - ) - return to_return - - return _go - - def _lazy_none_clause( - self, - reverse_direction: bool = False, - adapt_source: Optional[_CoreAdapterProto] = None, - ) -> ColumnElement[bool]: - if not reverse_direction: - criterion, bind_to_col = ( - self._lazy_strategy._lazywhere, - self._lazy_strategy._bind_to_col, - ) - else: - criterion, bind_to_col = ( - self._lazy_strategy._rev_lazywhere, - self._lazy_strategy._rev_bind_to_col, - ) - - criterion = adapt_criterion_to_null(criterion, bind_to_col) - - if adapt_source: - criterion = adapt_source(criterion) - return criterion - - def __str__(self) -> str: - return str(self.parent.class_.__name__) + "." + self.key - - def merge( - self, - session: Session, - source_state: InstanceState[Any], - source_dict: _InstanceDict, - dest_state: InstanceState[Any], - dest_dict: _InstanceDict, - load: bool, - _recursive: Dict[Any, object], - _resolve_conflict_map: Dict[_IdentityKeyType[Any], object], - ) -> None: - if load: - for r in self._reverse_property: - if (source_state, r) in _recursive: - return - - if "merge" not in self._cascade: - return - - if self.key not in source_dict: - return - - if self.uselist: - impl = source_state.get_impl(self.key) - - assert is_has_collection_adapter(impl) - instances_iterable = impl.get_collection(source_state, source_dict) - - # if this is a CollectionAttributeImpl, then empty should - # be False, otherwise "self.key in source_dict" should not be - # True - assert not instances_iterable.empty if impl.collection else True - - if load: - # for a full merge, pre-load the destination collection, - # so that individual _merge of each item pulls from identity - # map for those already present. - # also assumes CollectionAttributeImpl behavior of loading - # "old" list in any case - dest_state.get_impl(self.key).get( - dest_state, dest_dict, passive=PassiveFlag.PASSIVE_MERGE - ) - - dest_list = [] - for current in instances_iterable: - current_state = attributes.instance_state(current) - current_dict = attributes.instance_dict(current) - _recursive[(current_state, self)] = True - obj = session._merge( - current_state, - current_dict, - load=load, - _recursive=_recursive, - _resolve_conflict_map=_resolve_conflict_map, - ) - if obj is not None: - dest_list.append(obj) - - if not load: - coll = attributes.init_state_collection( - dest_state, dest_dict, self.key - ) - for c in dest_list: - coll.append_without_event(c) - else: - dest_impl = dest_state.get_impl(self.key) - assert is_has_collection_adapter(dest_impl) - dest_impl.set( - dest_state, - dest_dict, - dest_list, - _adapt=False, - passive=PassiveFlag.PASSIVE_MERGE, - ) - else: - current = source_dict[self.key] - if current is not None: - current_state = attributes.instance_state(current) - current_dict = attributes.instance_dict(current) - _recursive[(current_state, self)] = True - obj = session._merge( - current_state, - current_dict, - load=load, - _recursive=_recursive, - _resolve_conflict_map=_resolve_conflict_map, - ) - else: - obj = None - - if not load: - dest_dict[self.key] = obj - else: - dest_state.get_impl(self.key).set( - dest_state, dest_dict, obj, None - ) - - def _value_as_iterable( - self, - state: InstanceState[_O], - dict_: _InstanceDict, - key: str, - passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, - ) -> Sequence[Tuple[InstanceState[_O], _O]]: - """Return a list of tuples (state, obj) for the given - key. - - returns an empty list if the value is None/empty/PASSIVE_NO_RESULT - """ - - impl = state.manager[key].impl - x = impl.get(state, dict_, passive=passive) - if x is LoaderCallableStatus.PASSIVE_NO_RESULT or x is None: - return [] - elif is_has_collection_adapter(impl): - return [ - (attributes.instance_state(o), o) - for o in impl.get_collection(state, dict_, x, passive=passive) - ] - else: - return [(attributes.instance_state(x), x)] - - def cascade_iterator( - self, - type_: str, - state: InstanceState[Any], - dict_: _InstanceDict, - visited_states: Set[InstanceState[Any]], - halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None, - ) -> Iterator[Tuple[Any, Mapper[Any], InstanceState[Any], _InstanceDict]]: - # assert type_ in self._cascade - - # only actively lazy load on the 'delete' cascade - if type_ != "delete" or self.passive_deletes: - passive = PassiveFlag.PASSIVE_NO_INITIALIZE - else: - passive = PassiveFlag.PASSIVE_OFF | PassiveFlag.NO_RAISE - - if type_ == "save-update": - tuples = state.manager[self.key].impl.get_all_pending(state, dict_) - else: - tuples = self._value_as_iterable( - state, dict_, self.key, passive=passive - ) - - skip_pending = ( - type_ == "refresh-expire" and "delete-orphan" not in self._cascade - ) - - for instance_state, c in tuples: - if instance_state in visited_states: - continue - - if c is None: - # would like to emit a warning here, but - # would not be consistent with collection.append(None) - # current behavior of silently skipping. - # see [ticket:2229] - continue - - assert instance_state is not None - instance_dict = attributes.instance_dict(c) - - if halt_on and halt_on(instance_state): - continue - - if skip_pending and not instance_state.key: - continue - - instance_mapper = instance_state.manager.mapper - - if not instance_mapper.isa(self.mapper.class_manager.mapper): - raise AssertionError( - "Attribute '%s' on class '%s' " - "doesn't handle objects " - "of type '%s'" - % (self.key, self.parent.class_, c.__class__) - ) - - visited_states.add(instance_state) - - yield c, instance_mapper, instance_state, instance_dict - - @property - def _effective_sync_backref(self) -> bool: - if self.viewonly: - return False - else: - return self.sync_backref is not False - - @staticmethod - def _check_sync_backref( - rel_a: RelationshipProperty[Any], rel_b: RelationshipProperty[Any] - ) -> None: - if rel_a.viewonly and rel_b.sync_backref: - raise sa_exc.InvalidRequestError( - "Relationship %s cannot specify sync_backref=True since %s " - "includes viewonly=True." % (rel_b, rel_a) - ) - if ( - rel_a.viewonly - and not rel_b.viewonly - and rel_b.sync_backref is not False - ): - rel_b.sync_backref = False - - def _add_reverse_property(self, key: str) -> None: - other = self.mapper.get_property(key, _configure_mappers=False) - if not isinstance(other, RelationshipProperty): - raise sa_exc.InvalidRequestError( - "back_populates on relationship '%s' refers to attribute '%s' " - "that is not a relationship. The back_populates parameter " - "should refer to the name of a relationship on the target " - "class." % (self, other) - ) - # viewonly and sync_backref cases - # 1. self.viewonly==True and other.sync_backref==True -> error - # 2. self.viewonly==True and other.viewonly==False and - # other.sync_backref==None -> warn sync_backref=False, set to False - self._check_sync_backref(self, other) - # 3. other.viewonly==True and self.sync_backref==True -> error - # 4. other.viewonly==True and self.viewonly==False and - # self.sync_backref==None -> warn sync_backref=False, set to False - self._check_sync_backref(other, self) - - self._reverse_property.add(other) - other._reverse_property.add(self) - - other._setup_entity() - - if not other.mapper.common_parent(self.parent): - raise sa_exc.ArgumentError( - "reverse_property %r on " - "relationship %s references relationship %s, which " - "does not reference mapper %s" - % (key, self, other, self.parent) - ) - - if ( - other._configure_started - and self.direction in (ONETOMANY, MANYTOONE) - and self.direction == other.direction - ): - raise sa_exc.ArgumentError( - "%s and back-reference %s are " - "both of the same direction %r. Did you mean to " - "set remote_side on the many-to-one side ?" - % (other, self, self.direction) - ) - - @util.memoized_property - def entity(self) -> _InternalEntityType[_T]: - """Return the target mapped entity, which is an inspect() of the - class or aliased class that is referenced by this - :class:`.RelationshipProperty`. - - """ - self.parent._check_configure() - return self.entity - - @util.memoized_property - def mapper(self) -> Mapper[_T]: - """Return the targeted :class:`_orm.Mapper` for this - :class:`.RelationshipProperty`. - - """ - return self.entity.mapper - - def do_init(self) -> None: - self._check_conflicts() - self._process_dependent_arguments() - self._setup_entity() - self._setup_registry_dependencies() - self._setup_join_conditions() - self._check_cascade_settings(self._cascade) - self._post_init() - self._generate_backref() - self._join_condition._warn_for_conflicting_sync_targets() - super().do_init() - self._lazy_strategy = cast( - "LazyLoader", self._get_strategy((("lazy", "select"),)) - ) - - def _setup_registry_dependencies(self) -> None: - self.parent.mapper.registry._set_depends_on( - self.entity.mapper.registry - ) - - def _process_dependent_arguments(self) -> None: - """Convert incoming configuration arguments to their - proper form. - - Callables are resolved, ORM annotations removed. - - """ - - # accept callables for other attributes which may require - # deferred initialization. This technique is used - # by declarative "string configs" and some recipes. - init_args = self._init_args - - for attr in ( - "order_by", - "primaryjoin", - "secondaryjoin", - "secondary", - "foreign_keys", - "remote_side", - ): - rel_arg = getattr(init_args, attr) - - rel_arg._resolve_against_registry(self._clsregistry_resolvers[1]) - - # remove "annotations" which are present if mapped class - # descriptors are used to create the join expression. - for attr in "primaryjoin", "secondaryjoin": - rel_arg = getattr(init_args, attr) - val = rel_arg.resolved - if val is not None: - rel_arg.resolved = _orm_deannotate( - coercions.expect( - roles.ColumnArgumentRole, val, argname=attr - ) - ) - - secondary = init_args.secondary.resolved - if secondary is not None and _is_mapped_class(secondary): - raise sa_exc.ArgumentError( - "secondary argument %s passed to to relationship() %s must " - "be a Table object or other FROM clause; can't send a mapped " - "class directly as rows in 'secondary' are persisted " - "independently of a class that is mapped " - "to that same table." % (secondary, self) - ) - - # ensure expressions in self.order_by, foreign_keys, - # remote_side are all columns, not strings. - if ( - init_args.order_by.resolved is not False - and init_args.order_by.resolved is not None - ): - self.order_by = tuple( - coercions.expect( - roles.ColumnArgumentRole, x, argname="order_by" - ) - for x in util.to_list(init_args.order_by.resolved) - ) - else: - self.order_by = False - - self._user_defined_foreign_keys = util.column_set( - coercions.expect( - roles.ColumnArgumentRole, x, argname="foreign_keys" - ) - for x in util.to_column_set(init_args.foreign_keys.resolved) - ) - - self.remote_side = util.column_set( - coercions.expect( - roles.ColumnArgumentRole, x, argname="remote_side" - ) - for x in util.to_column_set(init_args.remote_side.resolved) - ) - - def declarative_scan( - self, - decl_scan: _ClassScanMapperConfig, - registry: _RegistryType, - cls: Type[Any], - originating_module: Optional[str], - key: str, - mapped_container: Optional[Type[Mapped[Any]]], - annotation: Optional[_AnnotationScanType], - extracted_mapped_annotation: Optional[_AnnotationScanType], - is_dataclass_field: bool, - ) -> None: - argument = extracted_mapped_annotation - - if extracted_mapped_annotation is None: - if self.argument is None: - self._raise_for_required(key, cls) - else: - return - - argument = extracted_mapped_annotation - assert originating_module is not None - - if mapped_container is not None: - is_write_only = issubclass(mapped_container, WriteOnlyMapped) - is_dynamic = issubclass(mapped_container, DynamicMapped) - if is_write_only: - self.lazy = "write_only" - self.strategy_key = (("lazy", self.lazy),) - elif is_dynamic: - self.lazy = "dynamic" - self.strategy_key = (("lazy", self.lazy),) - else: - is_write_only = is_dynamic = False - - argument = de_optionalize_union_types(argument) - - if hasattr(argument, "__origin__"): - arg_origin = argument.__origin__ - if isinstance(arg_origin, type) and issubclass( - arg_origin, abc.Collection - ): - if self.collection_class is None: - if _py_inspect.isabstract(arg_origin): - raise sa_exc.ArgumentError( - f"Collection annotation type {arg_origin} cannot " - "be instantiated; please provide an explicit " - "'collection_class' parameter " - "(e.g. list, set, etc.) to the " - "relationship() function to accompany this " - "annotation" - ) - - self.collection_class = arg_origin - - elif not is_write_only and not is_dynamic: - self.uselist = False - - if argument.__args__: # type: ignore - if isinstance(arg_origin, type) and issubclass( - arg_origin, typing.Mapping - ): - type_arg = argument.__args__[-1] # type: ignore - else: - type_arg = argument.__args__[0] # type: ignore - if hasattr(type_arg, "__forward_arg__"): - str_argument = type_arg.__forward_arg__ - - argument = resolve_name_to_real_class_name( - str_argument, originating_module - ) - else: - argument = type_arg - else: - raise sa_exc.ArgumentError( - f"Generic alias {argument} requires an argument" - ) - elif hasattr(argument, "__forward_arg__"): - argument = argument.__forward_arg__ - - argument = resolve_name_to_real_class_name( - argument, originating_module - ) - - if ( - self.collection_class is None - and not is_write_only - and not is_dynamic - ): - self.uselist = False - - # ticket #8759 - # if a lead argument was given to relationship(), like - # `relationship("B")`, use that, don't replace it with class we - # found in the annotation. The declarative_scan() method call here is - # still useful, as we continue to derive collection type and do - # checking of the annotation in any case. - if self.argument is None: - self.argument = cast("_RelationshipArgumentType[_T]", argument) - - @util.preload_module("sqlalchemy.orm.mapper") - def _setup_entity(self, __argument: Any = None) -> None: - if "entity" in self.__dict__: - return - - mapperlib = util.preloaded.orm_mapper - - if __argument: - argument = __argument - else: - argument = self.argument - - resolved_argument: _ExternalEntityType[Any] - - if isinstance(argument, str): - # we might want to cleanup clsregistry API to make this - # more straightforward - resolved_argument = cast( - "_ExternalEntityType[Any]", - self._clsregistry_resolve_name(argument)(), - ) - elif callable(argument) and not isinstance( - argument, (type, mapperlib.Mapper) - ): - resolved_argument = argument() - else: - resolved_argument = argument - - entity: _InternalEntityType[Any] - - if isinstance(resolved_argument, type): - entity = class_mapper(resolved_argument, configure=False) - else: - try: - entity = inspect(resolved_argument) - except sa_exc.NoInspectionAvailable: - entity = None # type: ignore - - if not hasattr(entity, "mapper"): - raise sa_exc.ArgumentError( - "relationship '%s' expects " - "a class or a mapper argument (received: %s)" - % (self.key, type(resolved_argument)) - ) - - self.entity = entity - self.target = self.entity.persist_selectable - - def _setup_join_conditions(self) -> None: - self._join_condition = jc = JoinCondition( - parent_persist_selectable=self.parent.persist_selectable, - child_persist_selectable=self.entity.persist_selectable, - parent_local_selectable=self.parent.local_table, - child_local_selectable=self.entity.local_table, - primaryjoin=self._init_args.primaryjoin.resolved, - secondary=self._init_args.secondary.resolved, - secondaryjoin=self._init_args.secondaryjoin.resolved, - parent_equivalents=self.parent._equivalent_columns, - child_equivalents=self.mapper._equivalent_columns, - consider_as_foreign_keys=self._user_defined_foreign_keys, - local_remote_pairs=self.local_remote_pairs, - remote_side=self.remote_side, - self_referential=self._is_self_referential, - prop=self, - support_sync=not self.viewonly, - can_be_synced_fn=self._columns_are_mapped, - ) - self.primaryjoin = jc.primaryjoin - self.secondaryjoin = jc.secondaryjoin - self.secondary = jc.secondary - self.direction = jc.direction - self.local_remote_pairs = jc.local_remote_pairs - self.remote_side = jc.remote_columns - self.local_columns = jc.local_columns - self.synchronize_pairs = jc.synchronize_pairs - self._calculated_foreign_keys = jc.foreign_key_columns - self.secondary_synchronize_pairs = jc.secondary_synchronize_pairs - - @property - def _clsregistry_resolve_arg( - self, - ) -> Callable[[str, bool], _class_resolver]: - return self._clsregistry_resolvers[1] - - @property - def _clsregistry_resolve_name( - self, - ) -> Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]]: - return self._clsregistry_resolvers[0] - - @util.memoized_property - @util.preload_module("sqlalchemy.orm.clsregistry") - def _clsregistry_resolvers( - self, - ) -> Tuple[ - Callable[[str], Callable[[], Union[Type[Any], Table, _ModNS]]], - Callable[[str, bool], _class_resolver], - ]: - _resolver = util.preloaded.orm_clsregistry._resolver - - return _resolver(self.parent.class_, self) - - def _check_conflicts(self) -> None: - """Test that this relationship is legal, warn about - inheritance conflicts.""" - if self.parent.non_primary and not class_mapper( - self.parent.class_, configure=False - ).has_property(self.key): - raise sa_exc.ArgumentError( - "Attempting to assign a new " - "relationship '%s' to a non-primary mapper on " - "class '%s'. New relationships can only be added " - "to the primary mapper, i.e. the very first mapper " - "created for class '%s' " - % ( - self.key, - self.parent.class_.__name__, - self.parent.class_.__name__, - ) - ) - - @property - def cascade(self) -> CascadeOptions: - """Return the current cascade setting for this - :class:`.RelationshipProperty`. - """ - return self._cascade - - @cascade.setter - def cascade(self, cascade: Union[str, CascadeOptions]) -> None: - self._set_cascade(cascade) - - def _set_cascade(self, cascade_arg: Union[str, CascadeOptions]) -> None: - cascade = CascadeOptions(cascade_arg) - - if self.viewonly: - cascade = CascadeOptions( - cascade.intersection(CascadeOptions._viewonly_cascades) - ) - - if "mapper" in self.__dict__: - self._check_cascade_settings(cascade) - self._cascade = cascade - - if self._dependency_processor: - self._dependency_processor.cascade = cascade - - def _check_cascade_settings(self, cascade: CascadeOptions) -> None: - if ( - cascade.delete_orphan - and not self.single_parent - and (self.direction is MANYTOMANY or self.direction is MANYTOONE) - ): - raise sa_exc.ArgumentError( - "For %(direction)s relationship %(rel)s, delete-orphan " - "cascade is normally " - 'configured only on the "one" side of a one-to-many ' - "relationship, " - 'and not on the "many" side of a many-to-one or many-to-many ' - "relationship. " - "To force this relationship to allow a particular " - '"%(relatedcls)s" object to be referenced by only ' - 'a single "%(clsname)s" object at a time via the ' - "%(rel)s relationship, which " - "would allow " - "delete-orphan cascade to take place in this direction, set " - "the single_parent=True flag." - % { - "rel": self, - "direction": ( - "many-to-one" - if self.direction is MANYTOONE - else "many-to-many" - ), - "clsname": self.parent.class_.__name__, - "relatedcls": self.mapper.class_.__name__, - }, - code="bbf0", - ) - - if self.passive_deletes == "all" and ( - "delete" in cascade or "delete-orphan" in cascade - ): - raise sa_exc.ArgumentError( - "On %s, can't set passive_deletes='all' in conjunction " - "with 'delete' or 'delete-orphan' cascade" % self - ) - - if cascade.delete_orphan: - self.mapper.primary_mapper()._delete_orphans.append( - (self.key, self.parent.class_) - ) - - def _persists_for(self, mapper: Mapper[Any]) -> bool: - """Return True if this property will persist values on behalf - of the given mapper. - - """ - - return ( - self.key in mapper.relationships - and mapper.relationships[self.key] is self - ) - - def _columns_are_mapped(self, *cols: ColumnElement[Any]) -> bool: - """Return True if all columns in the given collection are - mapped by the tables referenced by this :class:`.RelationshipProperty`. - - """ - - secondary = self._init_args.secondary.resolved - for c in cols: - if secondary is not None and secondary.c.contains_column(c): - continue - if not self.parent.persist_selectable.c.contains_column( - c - ) and not self.target.c.contains_column(c): - return False - return True - - def _generate_backref(self) -> None: - """Interpret the 'backref' instruction to create a - :func:`_orm.relationship` complementary to this one.""" - - if self.parent.non_primary: - return - if self.backref is not None and not self.back_populates: - kwargs: Dict[str, Any] - if isinstance(self.backref, str): - backref_key, kwargs = self.backref, {} - else: - backref_key, kwargs = self.backref - mapper = self.mapper.primary_mapper() - - if not mapper.concrete: - check = set(mapper.iterate_to_root()).union( - mapper.self_and_descendants - ) - for m in check: - if m.has_property(backref_key) and not m.concrete: - raise sa_exc.ArgumentError( - "Error creating backref " - "'%s' on relationship '%s': property of that " - "name exists on mapper '%s'" - % (backref_key, self, m) - ) - - # determine primaryjoin/secondaryjoin for the - # backref. Use the one we had, so that - # a custom join doesn't have to be specified in - # both directions. - if self.secondary is not None: - # for many to many, just switch primaryjoin/ - # secondaryjoin. use the annotated - # pj/sj on the _join_condition. - pj = kwargs.pop( - "primaryjoin", - self._join_condition.secondaryjoin_minus_local, - ) - sj = kwargs.pop( - "secondaryjoin", - self._join_condition.primaryjoin_minus_local, - ) - else: - pj = kwargs.pop( - "primaryjoin", - self._join_condition.primaryjoin_reverse_remote, - ) - sj = kwargs.pop("secondaryjoin", None) - if sj: - raise sa_exc.InvalidRequestError( - "Can't assign 'secondaryjoin' on a backref " - "against a non-secondary relationship." - ) - - foreign_keys = kwargs.pop( - "foreign_keys", self._user_defined_foreign_keys - ) - parent = self.parent.primary_mapper() - kwargs.setdefault("viewonly", self.viewonly) - kwargs.setdefault("post_update", self.post_update) - kwargs.setdefault("passive_updates", self.passive_updates) - kwargs.setdefault("sync_backref", self.sync_backref) - self.back_populates = backref_key - relationship = RelationshipProperty( - parent, - self.secondary, - primaryjoin=pj, - secondaryjoin=sj, - foreign_keys=foreign_keys, - back_populates=self.key, - **kwargs, - ) - mapper._configure_property( - backref_key, relationship, warn_for_existing=True - ) - - if self.back_populates: - self._add_reverse_property(self.back_populates) - - @util.preload_module("sqlalchemy.orm.dependency") - def _post_init(self) -> None: - dependency = util.preloaded.orm_dependency - - if self.uselist is None: - self.uselist = self.direction is not MANYTOONE - if not self.viewonly: - self._dependency_processor = ( # type: ignore - dependency.DependencyProcessor.from_relationship - )(self) - - @util.memoized_property - def _use_get(self) -> bool: - """memoize the 'use_get' attribute of this RelationshipLoader's - lazyloader.""" - - strategy = self._lazy_strategy - return strategy.use_get - - @util.memoized_property - def _is_self_referential(self) -> bool: - return self.mapper.common_parent(self.parent) - - def _create_joins( - self, - source_polymorphic: bool = False, - source_selectable: Optional[FromClause] = None, - dest_selectable: Optional[FromClause] = None, - of_type_entity: Optional[_InternalEntityType[Any]] = None, - alias_secondary: bool = False, - extra_criteria: Tuple[ColumnElement[bool], ...] = (), - ) -> Tuple[ - ColumnElement[bool], - Optional[ColumnElement[bool]], - FromClause, - FromClause, - Optional[FromClause], - Optional[ClauseAdapter], - ]: - aliased = False - - if alias_secondary and self.secondary is not None: - aliased = True - - if source_selectable is None: - if source_polymorphic and self.parent.with_polymorphic: - source_selectable = self.parent._with_polymorphic_selectable - - if of_type_entity: - dest_mapper = of_type_entity.mapper - if dest_selectable is None: - dest_selectable = of_type_entity.selectable - aliased = True - else: - dest_mapper = self.mapper - - if dest_selectable is None: - dest_selectable = self.entity.selectable - if self.mapper.with_polymorphic: - aliased = True - - if self._is_self_referential and source_selectable is None: - dest_selectable = dest_selectable._anonymous_fromclause() - aliased = True - elif ( - dest_selectable is not self.mapper._with_polymorphic_selectable - or self.mapper.with_polymorphic - ): - aliased = True - - single_crit = dest_mapper._single_table_criterion - aliased = aliased or ( - source_selectable is not None - and ( - source_selectable - is not self.parent._with_polymorphic_selectable - or source_selectable._is_subquery - ) - ) - - ( - primaryjoin, - secondaryjoin, - secondary, - target_adapter, - dest_selectable, - ) = self._join_condition.join_targets( - source_selectable, - dest_selectable, - aliased, - single_crit, - extra_criteria, - ) - if source_selectable is None: - source_selectable = self.parent.local_table - if dest_selectable is None: - dest_selectable = self.entity.local_table - return ( - primaryjoin, - secondaryjoin, - source_selectable, - dest_selectable, - secondary, - target_adapter, - ) - - -def _annotate_columns(element: _CE, annotations: _AnnotationDict) -> _CE: - def clone(elem: _CE) -> _CE: - if isinstance(elem, expression.ColumnClause): - elem = elem._annotate(annotations.copy()) # type: ignore - elem._copy_internals(clone=clone) - return elem - - if element is not None: - element = clone(element) - clone = None # type: ignore # remove gc cycles - return element - - -class JoinCondition: - primaryjoin_initial: Optional[ColumnElement[bool]] - primaryjoin: ColumnElement[bool] - secondaryjoin: Optional[ColumnElement[bool]] - secondary: Optional[FromClause] - prop: RelationshipProperty[Any] - - synchronize_pairs: _ColumnPairs - secondary_synchronize_pairs: _ColumnPairs - direction: RelationshipDirection - - parent_persist_selectable: FromClause - child_persist_selectable: FromClause - parent_local_selectable: FromClause - child_local_selectable: FromClause - - _local_remote_pairs: Optional[_ColumnPairs] - - def __init__( - self, - parent_persist_selectable: FromClause, - child_persist_selectable: FromClause, - parent_local_selectable: FromClause, - child_local_selectable: FromClause, - *, - primaryjoin: Optional[ColumnElement[bool]] = None, - secondary: Optional[FromClause] = None, - secondaryjoin: Optional[ColumnElement[bool]] = None, - parent_equivalents: Optional[_EquivalentColumnMap] = None, - child_equivalents: Optional[_EquivalentColumnMap] = None, - consider_as_foreign_keys: Any = None, - local_remote_pairs: Optional[_ColumnPairs] = None, - remote_side: Any = None, - self_referential: Any = False, - prop: RelationshipProperty[Any], - support_sync: bool = True, - can_be_synced_fn: Callable[..., bool] = lambda *c: True, - ): - self.parent_persist_selectable = parent_persist_selectable - self.parent_local_selectable = parent_local_selectable - self.child_persist_selectable = child_persist_selectable - self.child_local_selectable = child_local_selectable - self.parent_equivalents = parent_equivalents - self.child_equivalents = child_equivalents - self.primaryjoin_initial = primaryjoin - self.secondaryjoin = secondaryjoin - self.secondary = secondary - self.consider_as_foreign_keys = consider_as_foreign_keys - self._local_remote_pairs = local_remote_pairs - self._remote_side = remote_side - self.prop = prop - self.self_referential = self_referential - self.support_sync = support_sync - self.can_be_synced_fn = can_be_synced_fn - - self._determine_joins() - assert self.primaryjoin is not None - - self._sanitize_joins() - self._annotate_fks() - self._annotate_remote() - self._annotate_local() - self._annotate_parentmapper() - self._setup_pairs() - self._check_foreign_cols(self.primaryjoin, True) - if self.secondaryjoin is not None: - self._check_foreign_cols(self.secondaryjoin, False) - self._determine_direction() - self._check_remote_side() - self._log_joins() - - def _log_joins(self) -> None: - log = self.prop.logger - log.info("%s setup primary join %s", self.prop, self.primaryjoin) - log.info("%s setup secondary join %s", self.prop, self.secondaryjoin) - log.info( - "%s synchronize pairs [%s]", - self.prop, - ",".join( - "(%s => %s)" % (l, r) for (l, r) in self.synchronize_pairs - ), - ) - log.info( - "%s secondary synchronize pairs [%s]", - self.prop, - ",".join( - "(%s => %s)" % (l, r) - for (l, r) in self.secondary_synchronize_pairs or [] - ), - ) - log.info( - "%s local/remote pairs [%s]", - self.prop, - ",".join( - "(%s / %s)" % (l, r) for (l, r) in self.local_remote_pairs - ), - ) - log.info( - "%s remote columns [%s]", - self.prop, - ",".join("%s" % col for col in self.remote_columns), - ) - log.info( - "%s local columns [%s]", - self.prop, - ",".join("%s" % col for col in self.local_columns), - ) - log.info("%s relationship direction %s", self.prop, self.direction) - - def _sanitize_joins(self) -> None: - """remove the parententity annotation from our join conditions which - can leak in here based on some declarative patterns and maybe others. - - "parentmapper" is relied upon both by the ORM evaluator as well as - the use case in _join_fixture_inh_selfref_w_entity - that relies upon it being present, see :ticket:`3364`. - - """ - - self.primaryjoin = _deep_deannotate( - self.primaryjoin, values=("parententity", "proxy_key") - ) - if self.secondaryjoin is not None: - self.secondaryjoin = _deep_deannotate( - self.secondaryjoin, values=("parententity", "proxy_key") - ) - - def _determine_joins(self) -> None: - """Determine the 'primaryjoin' and 'secondaryjoin' attributes, - if not passed to the constructor already. - - This is based on analysis of the foreign key relationships - between the parent and target mapped selectables. - - """ - if self.secondaryjoin is not None and self.secondary is None: - raise sa_exc.ArgumentError( - "Property %s specified with secondary " - "join condition but " - "no secondary argument" % self.prop - ) - - # find a join between the given mapper's mapped table and - # the given table. will try the mapper's local table first - # for more specificity, then if not found will try the more - # general mapped table, which in the case of inheritance is - # a join. - try: - consider_as_foreign_keys = self.consider_as_foreign_keys or None - if self.secondary is not None: - if self.secondaryjoin is None: - self.secondaryjoin = join_condition( - self.child_persist_selectable, - self.secondary, - a_subset=self.child_local_selectable, - consider_as_foreign_keys=consider_as_foreign_keys, - ) - if self.primaryjoin_initial is None: - self.primaryjoin = join_condition( - self.parent_persist_selectable, - self.secondary, - a_subset=self.parent_local_selectable, - consider_as_foreign_keys=consider_as_foreign_keys, - ) - else: - self.primaryjoin = self.primaryjoin_initial - else: - if self.primaryjoin_initial is None: - self.primaryjoin = join_condition( - self.parent_persist_selectable, - self.child_persist_selectable, - a_subset=self.parent_local_selectable, - consider_as_foreign_keys=consider_as_foreign_keys, - ) - else: - self.primaryjoin = self.primaryjoin_initial - except sa_exc.NoForeignKeysError as nfe: - if self.secondary is not None: - raise sa_exc.NoForeignKeysError( - "Could not determine join " - "condition between parent/child tables on " - "relationship %s - there are no foreign keys " - "linking these tables via secondary table '%s'. " - "Ensure that referencing columns are associated " - "with a ForeignKey or ForeignKeyConstraint, or " - "specify 'primaryjoin' and 'secondaryjoin' " - "expressions." % (self.prop, self.secondary) - ) from nfe - else: - raise sa_exc.NoForeignKeysError( - "Could not determine join " - "condition between parent/child tables on " - "relationship %s - there are no foreign keys " - "linking these tables. " - "Ensure that referencing columns are associated " - "with a ForeignKey or ForeignKeyConstraint, or " - "specify a 'primaryjoin' expression." % self.prop - ) from nfe - except sa_exc.AmbiguousForeignKeysError as afe: - if self.secondary is not None: - raise sa_exc.AmbiguousForeignKeysError( - "Could not determine join " - "condition between parent/child tables on " - "relationship %s - there are multiple foreign key " - "paths linking the tables via secondary table '%s'. " - "Specify the 'foreign_keys' " - "argument, providing a list of those columns which " - "should be counted as containing a foreign key " - "reference from the secondary table to each of the " - "parent and child tables." % (self.prop, self.secondary) - ) from afe - else: - raise sa_exc.AmbiguousForeignKeysError( - "Could not determine join " - "condition between parent/child tables on " - "relationship %s - there are multiple foreign key " - "paths linking the tables. Specify the " - "'foreign_keys' argument, providing a list of those " - "columns which should be counted as containing a " - "foreign key reference to the parent table." % self.prop - ) from afe - - @property - def primaryjoin_minus_local(self) -> ColumnElement[bool]: - return _deep_deannotate(self.primaryjoin, values=("local", "remote")) - - @property - def secondaryjoin_minus_local(self) -> ColumnElement[bool]: - assert self.secondaryjoin is not None - return _deep_deannotate(self.secondaryjoin, values=("local", "remote")) - - @util.memoized_property - def primaryjoin_reverse_remote(self) -> ColumnElement[bool]: - """Return the primaryjoin condition suitable for the - "reverse" direction. - - If the primaryjoin was delivered here with pre-existing - "remote" annotations, the local/remote annotations - are reversed. Otherwise, the local/remote annotations - are removed. - - """ - if self._has_remote_annotations: - - def replace(element: _CE, **kw: Any) -> Optional[_CE]: - if "remote" in element._annotations: - v = dict(element._annotations) - del v["remote"] - v["local"] = True - return element._with_annotations(v) - elif "local" in element._annotations: - v = dict(element._annotations) - del v["local"] - v["remote"] = True - return element._with_annotations(v) - - return None - - return visitors.replacement_traverse(self.primaryjoin, {}, replace) - else: - if self._has_foreign_annotations: - # TODO: coverage - return _deep_deannotate( - self.primaryjoin, values=("local", "remote") - ) - else: - return _deep_deannotate(self.primaryjoin) - - def _has_annotation(self, clause: ClauseElement, annotation: str) -> bool: - for col in visitors.iterate(clause, {}): - if annotation in col._annotations: - return True - else: - return False - - @util.memoized_property - def _has_foreign_annotations(self) -> bool: - return self._has_annotation(self.primaryjoin, "foreign") - - @util.memoized_property - def _has_remote_annotations(self) -> bool: - return self._has_annotation(self.primaryjoin, "remote") - - def _annotate_fks(self) -> None: - """Annotate the primaryjoin and secondaryjoin - structures with 'foreign' annotations marking columns - considered as foreign. - - """ - if self._has_foreign_annotations: - return - - if self.consider_as_foreign_keys: - self._annotate_from_fk_list() - else: - self._annotate_present_fks() - - def _annotate_from_fk_list(self) -> None: - def check_fk(element: _CE, **kw: Any) -> Optional[_CE]: - if element in self.consider_as_foreign_keys: - return element._annotate({"foreign": True}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, check_fk - ) - if self.secondaryjoin is not None: - self.secondaryjoin = visitors.replacement_traverse( - self.secondaryjoin, {}, check_fk - ) - - def _annotate_present_fks(self) -> None: - if self.secondary is not None: - secondarycols = util.column_set(self.secondary.c) - else: - secondarycols = set() - - def is_foreign( - a: ColumnElement[Any], b: ColumnElement[Any] - ) -> Optional[ColumnElement[Any]]: - if isinstance(a, schema.Column) and isinstance(b, schema.Column): - if a.references(b): - return a - elif b.references(a): - return b - - if secondarycols: - if a in secondarycols and b not in secondarycols: - return a - elif b in secondarycols and a not in secondarycols: - return b - - return None - - def visit_binary(binary: BinaryExpression[Any]) -> None: - if not isinstance( - binary.left, sql.ColumnElement - ) or not isinstance(binary.right, sql.ColumnElement): - return - - if ( - "foreign" not in binary.left._annotations - and "foreign" not in binary.right._annotations - ): - col = is_foreign(binary.left, binary.right) - if col is not None: - if col.compare(binary.left): - binary.left = binary.left._annotate({"foreign": True}) - elif col.compare(binary.right): - binary.right = binary.right._annotate( - {"foreign": True} - ) - - self.primaryjoin = visitors.cloned_traverse( - self.primaryjoin, {}, {"binary": visit_binary} - ) - if self.secondaryjoin is not None: - self.secondaryjoin = visitors.cloned_traverse( - self.secondaryjoin, {}, {"binary": visit_binary} - ) - - def _refers_to_parent_table(self) -> bool: - """Return True if the join condition contains column - comparisons where both columns are in both tables. - - """ - pt = self.parent_persist_selectable - mt = self.child_persist_selectable - result = False - - def visit_binary(binary: BinaryExpression[Any]) -> None: - nonlocal result - c, f = binary.left, binary.right - if ( - isinstance(c, expression.ColumnClause) - and isinstance(f, expression.ColumnClause) - and pt.is_derived_from(c.table) - and pt.is_derived_from(f.table) - and mt.is_derived_from(c.table) - and mt.is_derived_from(f.table) - ): - result = True - - visitors.traverse(self.primaryjoin, {}, {"binary": visit_binary}) - return result - - def _tables_overlap(self) -> bool: - """Return True if parent/child tables have some overlap.""" - - return selectables_overlap( - self.parent_persist_selectable, self.child_persist_selectable - ) - - def _annotate_remote(self) -> None: - """Annotate the primaryjoin and secondaryjoin - structures with 'remote' annotations marking columns - considered as part of the 'remote' side. - - """ - if self._has_remote_annotations: - return - - if self.secondary is not None: - self._annotate_remote_secondary() - elif self._local_remote_pairs or self._remote_side: - self._annotate_remote_from_args() - elif self._refers_to_parent_table(): - self._annotate_selfref( - lambda col: "foreign" in col._annotations, False - ) - elif self._tables_overlap(): - self._annotate_remote_with_overlap() - else: - self._annotate_remote_distinct_selectables() - - def _annotate_remote_secondary(self) -> None: - """annotate 'remote' in primaryjoin, secondaryjoin - when 'secondary' is present. - - """ - - assert self.secondary is not None - fixed_secondary = self.secondary - - def repl(element: _CE, **kw: Any) -> Optional[_CE]: - if fixed_secondary.c.contains_column(element): - return element._annotate({"remote": True}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, repl - ) - - assert self.secondaryjoin is not None - self.secondaryjoin = visitors.replacement_traverse( - self.secondaryjoin, {}, repl - ) - - def _annotate_selfref( - self, fn: Callable[[ColumnElement[Any]], bool], remote_side_given: bool - ) -> None: - """annotate 'remote' in primaryjoin, secondaryjoin - when the relationship is detected as self-referential. - - """ - - def visit_binary(binary: BinaryExpression[Any]) -> None: - equated = binary.left.compare(binary.right) - if isinstance(binary.left, expression.ColumnClause) and isinstance( - binary.right, expression.ColumnClause - ): - # assume one to many - FKs are "remote" - if fn(binary.left): - binary.left = binary.left._annotate({"remote": True}) - if fn(binary.right) and not equated: - binary.right = binary.right._annotate({"remote": True}) - elif not remote_side_given: - self._warn_non_column_elements() - - self.primaryjoin = visitors.cloned_traverse( - self.primaryjoin, {}, {"binary": visit_binary} - ) - - def _annotate_remote_from_args(self) -> None: - """annotate 'remote' in primaryjoin, secondaryjoin - when the 'remote_side' or '_local_remote_pairs' - arguments are used. - - """ - if self._local_remote_pairs: - if self._remote_side: - raise sa_exc.ArgumentError( - "remote_side argument is redundant " - "against more detailed _local_remote_side " - "argument." - ) - - remote_side = [r for (l, r) in self._local_remote_pairs] - else: - remote_side = self._remote_side - - if self._refers_to_parent_table(): - self._annotate_selfref(lambda col: col in remote_side, True) - else: - - def repl(element: _CE, **kw: Any) -> Optional[_CE]: - # use set() to avoid generating ``__eq__()`` expressions - # against each element - if element in set(remote_side): - return element._annotate({"remote": True}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, repl - ) - - def _annotate_remote_with_overlap(self) -> None: - """annotate 'remote' in primaryjoin, secondaryjoin - when the parent/child tables have some set of - tables in common, though is not a fully self-referential - relationship. - - """ - - def visit_binary(binary: BinaryExpression[Any]) -> None: - binary.left, binary.right = proc_left_right( - binary.left, binary.right - ) - binary.right, binary.left = proc_left_right( - binary.right, binary.left - ) - - check_entities = ( - self.prop is not None and self.prop.mapper is not self.prop.parent - ) - - def proc_left_right( - left: ColumnElement[Any], right: ColumnElement[Any] - ) -> Tuple[ColumnElement[Any], ColumnElement[Any]]: - if isinstance(left, expression.ColumnClause) and isinstance( - right, expression.ColumnClause - ): - if self.child_persist_selectable.c.contains_column( - right - ) and self.parent_persist_selectable.c.contains_column(left): - right = right._annotate({"remote": True}) - elif ( - check_entities - and right._annotations.get("parentmapper") is self.prop.mapper - ): - right = right._annotate({"remote": True}) - elif ( - check_entities - and left._annotations.get("parentmapper") is self.prop.mapper - ): - left = left._annotate({"remote": True}) - else: - self._warn_non_column_elements() - - return left, right - - self.primaryjoin = visitors.cloned_traverse( - self.primaryjoin, {}, {"binary": visit_binary} - ) - - def _annotate_remote_distinct_selectables(self) -> None: - """annotate 'remote' in primaryjoin, secondaryjoin - when the parent/child tables are entirely - separate. - - """ - - def repl(element: _CE, **kw: Any) -> Optional[_CE]: - if self.child_persist_selectable.c.contains_column(element) and ( - not self.parent_local_selectable.c.contains_column(element) - or self.child_local_selectable.c.contains_column(element) - ): - return element._annotate({"remote": True}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, repl - ) - - def _warn_non_column_elements(self) -> None: - util.warn( - "Non-simple column elements in primary " - "join condition for property %s - consider using " - "remote() annotations to mark the remote side." % self.prop - ) - - def _annotate_local(self) -> None: - """Annotate the primaryjoin and secondaryjoin - structures with 'local' annotations. - - This annotates all column elements found - simultaneously in the parent table - and the join condition that don't have a - 'remote' annotation set up from - _annotate_remote() or user-defined. - - """ - if self._has_annotation(self.primaryjoin, "local"): - return - - if self._local_remote_pairs: - local_side = util.column_set( - [l for (l, r) in self._local_remote_pairs] - ) - else: - local_side = util.column_set(self.parent_persist_selectable.c) - - def locals_(element: _CE, **kw: Any) -> Optional[_CE]: - if "remote" not in element._annotations and element in local_side: - return element._annotate({"local": True}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, locals_ - ) - - def _annotate_parentmapper(self) -> None: - def parentmappers_(element: _CE, **kw: Any) -> Optional[_CE]: - if "remote" in element._annotations: - return element._annotate({"parentmapper": self.prop.mapper}) - elif "local" in element._annotations: - return element._annotate({"parentmapper": self.prop.parent}) - return None - - self.primaryjoin = visitors.replacement_traverse( - self.primaryjoin, {}, parentmappers_ - ) - - def _check_remote_side(self) -> None: - if not self.local_remote_pairs: - raise sa_exc.ArgumentError( - "Relationship %s could " - "not determine any unambiguous local/remote column " - "pairs based on join condition and remote_side " - "arguments. " - "Consider using the remote() annotation to " - "accurately mark those elements of the join " - "condition that are on the remote side of " - "the relationship." % (self.prop,) - ) - else: - not_target = util.column_set( - self.parent_persist_selectable.c - ).difference(self.child_persist_selectable.c) - - for _, rmt in self.local_remote_pairs: - if rmt in not_target: - util.warn( - "Expression %s is marked as 'remote', but these " - "column(s) are local to the local side. The " - "remote() annotation is needed only for a " - "self-referential relationship where both sides " - "of the relationship refer to the same tables." - % (rmt,) - ) - - def _check_foreign_cols( - self, join_condition: ColumnElement[bool], primary: bool - ) -> None: - """Check the foreign key columns collected and emit error - messages.""" - - can_sync = False - - foreign_cols = self._gather_columns_with_annotation( - join_condition, "foreign" - ) - - has_foreign = bool(foreign_cols) - - if primary: - can_sync = bool(self.synchronize_pairs) - else: - can_sync = bool(self.secondary_synchronize_pairs) - - if ( - self.support_sync - and can_sync - or (not self.support_sync and has_foreign) - ): - return - - # from here below is just determining the best error message - # to report. Check for a join condition using any operator - # (not just ==), perhaps they need to turn on "viewonly=True". - if self.support_sync and has_foreign and not can_sync: - err = ( - "Could not locate any simple equality expressions " - "involving locally mapped foreign key columns for " - "%s join condition " - "'%s' on relationship %s." - % ( - primary and "primary" or "secondary", - join_condition, - self.prop, - ) - ) - err += ( - " Ensure that referencing columns are associated " - "with a ForeignKey or ForeignKeyConstraint, or are " - "annotated in the join condition with the foreign() " - "annotation. To allow comparison operators other than " - "'==', the relationship can be marked as viewonly=True." - ) - - raise sa_exc.ArgumentError(err) - else: - err = ( - "Could not locate any relevant foreign key columns " - "for %s join condition '%s' on relationship %s." - % ( - primary and "primary" or "secondary", - join_condition, - self.prop, - ) - ) - err += ( - " Ensure that referencing columns are associated " - "with a ForeignKey or ForeignKeyConstraint, or are " - "annotated in the join condition with the foreign() " - "annotation." - ) - raise sa_exc.ArgumentError(err) - - def _determine_direction(self) -> None: - """Determine if this relationship is one to many, many to one, - many to many. - - """ - if self.secondaryjoin is not None: - self.direction = MANYTOMANY - else: - parentcols = util.column_set(self.parent_persist_selectable.c) - targetcols = util.column_set(self.child_persist_selectable.c) - - # fk collection which suggests ONETOMANY. - onetomany_fk = targetcols.intersection(self.foreign_key_columns) - - # fk collection which suggests MANYTOONE. - - manytoone_fk = parentcols.intersection(self.foreign_key_columns) - - if onetomany_fk and manytoone_fk: - # fks on both sides. test for overlap of local/remote - # with foreign key. - # we will gather columns directly from their annotations - # without deannotating, so that we can distinguish on a column - # that refers to itself. - - # 1. columns that are both remote and FK suggest - # onetomany. - onetomany_local = self._gather_columns_with_annotation( - self.primaryjoin, "remote", "foreign" - ) - - # 2. columns that are FK but are not remote (e.g. local) - # suggest manytoone. - manytoone_local = { - c - for c in self._gather_columns_with_annotation( - self.primaryjoin, "foreign" - ) - if "remote" not in c._annotations - } - - # 3. if both collections are present, remove columns that - # refer to themselves. This is for the case of - # and_(Me.id == Me.remote_id, Me.version == Me.version) - if onetomany_local and manytoone_local: - self_equated = self.remote_columns.intersection( - self.local_columns - ) - onetomany_local = onetomany_local.difference(self_equated) - manytoone_local = manytoone_local.difference(self_equated) - - # at this point, if only one or the other collection is - # present, we know the direction, otherwise it's still - # ambiguous. - - if onetomany_local and not manytoone_local: - self.direction = ONETOMANY - elif manytoone_local and not onetomany_local: - self.direction = MANYTOONE - else: - raise sa_exc.ArgumentError( - "Can't determine relationship" - " direction for relationship '%s' - foreign " - "key columns within the join condition are present " - "in both the parent and the child's mapped tables. " - "Ensure that only those columns referring " - "to a parent column are marked as foreign, " - "either via the foreign() annotation or " - "via the foreign_keys argument." % self.prop - ) - elif onetomany_fk: - self.direction = ONETOMANY - elif manytoone_fk: - self.direction = MANYTOONE - else: - raise sa_exc.ArgumentError( - "Can't determine relationship " - "direction for relationship '%s' - foreign " - "key columns are present in neither the parent " - "nor the child's mapped tables" % self.prop - ) - - def _deannotate_pairs( - self, collection: _ColumnPairIterable - ) -> _MutableColumnPairs: - """provide deannotation for the various lists of - pairs, so that using them in hashes doesn't incur - high-overhead __eq__() comparisons against - original columns mapped. - - """ - return [(x._deannotate(), y._deannotate()) for x, y in collection] - - def _setup_pairs(self) -> None: - sync_pairs: _MutableColumnPairs = [] - lrp: util.OrderedSet[Tuple[ColumnElement[Any], ColumnElement[Any]]] = ( - util.OrderedSet([]) - ) - secondary_sync_pairs: _MutableColumnPairs = [] - - def go( - joincond: ColumnElement[bool], - collection: _MutableColumnPairs, - ) -> None: - def visit_binary( - binary: BinaryExpression[Any], - left: ColumnElement[Any], - right: ColumnElement[Any], - ) -> None: - if ( - "remote" in right._annotations - and "remote" not in left._annotations - and self.can_be_synced_fn(left) - ): - lrp.add((left, right)) - elif ( - "remote" in left._annotations - and "remote" not in right._annotations - and self.can_be_synced_fn(right) - ): - lrp.add((right, left)) - if binary.operator is operators.eq and self.can_be_synced_fn( - left, right - ): - if "foreign" in right._annotations: - collection.append((left, right)) - elif "foreign" in left._annotations: - collection.append((right, left)) - - visit_binary_product(visit_binary, joincond) - - for joincond, collection in [ - (self.primaryjoin, sync_pairs), - (self.secondaryjoin, secondary_sync_pairs), - ]: - if joincond is None: - continue - go(joincond, collection) - - self.local_remote_pairs = self._deannotate_pairs(lrp) - self.synchronize_pairs = self._deannotate_pairs(sync_pairs) - self.secondary_synchronize_pairs = self._deannotate_pairs( - secondary_sync_pairs - ) - - _track_overlapping_sync_targets: weakref.WeakKeyDictionary[ - ColumnElement[Any], - weakref.WeakKeyDictionary[ - RelationshipProperty[Any], ColumnElement[Any] - ], - ] = weakref.WeakKeyDictionary() - - def _warn_for_conflicting_sync_targets(self) -> None: - if not self.support_sync: - return - - # we would like to detect if we are synchronizing any column - # pairs in conflict with another relationship that wishes to sync - # an entirely different column to the same target. This is a - # very rare edge case so we will try to minimize the memory/overhead - # impact of this check - for from_, to_ in [ - (from_, to_) for (from_, to_) in self.synchronize_pairs - ] + [ - (from_, to_) for (from_, to_) in self.secondary_synchronize_pairs - ]: - # save ourselves a ton of memory and overhead by only - # considering columns that are subject to a overlapping - # FK constraints at the core level. This condition can arise - # if multiple relationships overlap foreign() directly, but - # we're going to assume it's typically a ForeignKeyConstraint- - # level configuration that benefits from this warning. - - if to_ not in self._track_overlapping_sync_targets: - self._track_overlapping_sync_targets[to_] = ( - weakref.WeakKeyDictionary({self.prop: from_}) - ) - else: - other_props = [] - prop_to_from = self._track_overlapping_sync_targets[to_] - - for pr, fr_ in prop_to_from.items(): - if ( - not pr.mapper._dispose_called - and pr not in self.prop._reverse_property - and pr.key not in self.prop._overlaps - and self.prop.key not in pr._overlaps - # note: the "__*" symbol is used internally by - # SQLAlchemy as a general means of suppressing the - # overlaps warning for some extension cases, however - # this is not currently - # a publicly supported symbol and may change at - # any time. - and "__*" not in self.prop._overlaps - and "__*" not in pr._overlaps - and not self.prop.parent.is_sibling(pr.parent) - and not self.prop.mapper.is_sibling(pr.mapper) - and not self.prop.parent.is_sibling(pr.mapper) - and not self.prop.mapper.is_sibling(pr.parent) - and ( - self.prop.key != pr.key - or not self.prop.parent.common_parent(pr.parent) - ) - ): - other_props.append((pr, fr_)) - - if other_props: - util.warn( - "relationship '%s' will copy column %s to column %s, " - "which conflicts with relationship(s): %s. " - "If this is not the intention, consider if these " - "relationships should be linked with " - "back_populates, or if viewonly=True should be " - "applied to one or more if they are read-only. " - "For the less common case that foreign key " - "constraints are partially overlapping, the " - "orm.foreign() " - "annotation can be used to isolate the columns that " - "should be written towards. To silence this " - "warning, add the parameter 'overlaps=\"%s\"' to the " - "'%s' relationship." - % ( - self.prop, - from_, - to_, - ", ".join( - sorted( - "'%s' (copies %s to %s)" % (pr, fr_, to_) - for (pr, fr_) in other_props - ) - ), - ",".join(sorted(pr.key for pr, fr in other_props)), - self.prop, - ), - code="qzyx", - ) - self._track_overlapping_sync_targets[to_][self.prop] = from_ - - @util.memoized_property - def remote_columns(self) -> Set[ColumnElement[Any]]: - return self._gather_join_annotations("remote") - - @util.memoized_property - def local_columns(self) -> Set[ColumnElement[Any]]: - return self._gather_join_annotations("local") - - @util.memoized_property - def foreign_key_columns(self) -> Set[ColumnElement[Any]]: - return self._gather_join_annotations("foreign") - - def _gather_join_annotations( - self, annotation: str - ) -> Set[ColumnElement[Any]]: - s = set( - self._gather_columns_with_annotation(self.primaryjoin, annotation) - ) - if self.secondaryjoin is not None: - s.update( - self._gather_columns_with_annotation( - self.secondaryjoin, annotation - ) - ) - return {x._deannotate() for x in s} - - def _gather_columns_with_annotation( - self, clause: ColumnElement[Any], *annotation: Iterable[str] - ) -> Set[ColumnElement[Any]]: - annotation_set = set(annotation) - return { - cast(ColumnElement[Any], col) - for col in visitors.iterate(clause, {}) - if annotation_set.issubset(col._annotations) - } - - @util.memoized_property - def _secondary_lineage_set(self) -> FrozenSet[ColumnElement[Any]]: - if self.secondary is not None: - return frozenset( - itertools.chain(*[c.proxy_set for c in self.secondary.c]) - ) - else: - return util.EMPTY_SET - - def join_targets( - self, - source_selectable: Optional[FromClause], - dest_selectable: FromClause, - aliased: bool, - single_crit: Optional[ColumnElement[bool]] = None, - extra_criteria: Tuple[ColumnElement[bool], ...] = (), - ) -> Tuple[ - ColumnElement[bool], - Optional[ColumnElement[bool]], - Optional[FromClause], - Optional[ClauseAdapter], - FromClause, - ]: - """Given a source and destination selectable, create a - join between them. - - This takes into account aliasing the join clause - to reference the appropriate corresponding columns - in the target objects, as well as the extra child - criterion, equivalent column sets, etc. - - """ - # place a barrier on the destination such that - # replacement traversals won't ever dig into it. - # its internal structure remains fixed - # regardless of context. - dest_selectable = _shallow_annotate( - dest_selectable, {"no_replacement_traverse": True} - ) - - primaryjoin, secondaryjoin, secondary = ( - self.primaryjoin, - self.secondaryjoin, - self.secondary, - ) - - # adjust the join condition for single table inheritance, - # in the case that the join is to a subclass - # this is analogous to the - # "_adjust_for_single_table_inheritance()" method in Query. - - if single_crit is not None: - if secondaryjoin is not None: - secondaryjoin = secondaryjoin & single_crit - else: - primaryjoin = primaryjoin & single_crit - - if extra_criteria: - - def mark_exclude_cols( - elem: SupportsAnnotations, annotations: _AnnotationDict - ) -> SupportsAnnotations: - """note unrelated columns in the "extra criteria" as either - should be adapted or not adapted, even though they are not - part of our "local" or "remote" side. - - see #9779 for this case, as well as #11010 for a follow up - - """ - - parentmapper_for_element = elem._annotations.get( - "parentmapper", None - ) - - if ( - parentmapper_for_element is not self.prop.parent - and parentmapper_for_element is not self.prop.mapper - and elem not in self._secondary_lineage_set - ): - return _safe_annotate(elem, annotations) - else: - return elem - - extra_criteria = tuple( - _deep_annotate( - elem, - {"should_not_adapt": True}, - annotate_callable=mark_exclude_cols, - ) - for elem in extra_criteria - ) - - if secondaryjoin is not None: - secondaryjoin = secondaryjoin & sql.and_(*extra_criteria) - else: - primaryjoin = primaryjoin & sql.and_(*extra_criteria) - - if aliased: - if secondary is not None: - secondary = secondary._anonymous_fromclause(flat=True) - primary_aliasizer = ClauseAdapter( - secondary, - exclude_fn=_local_col_exclude, - ) - secondary_aliasizer = ClauseAdapter( - dest_selectable, equivalents=self.child_equivalents - ).chain(primary_aliasizer) - if source_selectable is not None: - primary_aliasizer = ClauseAdapter( - secondary, - exclude_fn=_local_col_exclude, - ).chain( - ClauseAdapter( - source_selectable, - equivalents=self.parent_equivalents, - ) - ) - - secondaryjoin = secondary_aliasizer.traverse(secondaryjoin) - else: - primary_aliasizer = ClauseAdapter( - dest_selectable, - exclude_fn=_local_col_exclude, - equivalents=self.child_equivalents, - ) - if source_selectable is not None: - primary_aliasizer.chain( - ClauseAdapter( - source_selectable, - exclude_fn=_remote_col_exclude, - equivalents=self.parent_equivalents, - ) - ) - secondary_aliasizer = None - - primaryjoin = primary_aliasizer.traverse(primaryjoin) - target_adapter = secondary_aliasizer or primary_aliasizer - target_adapter.exclude_fn = None - else: - target_adapter = None - return ( - primaryjoin, - secondaryjoin, - secondary, - target_adapter, - dest_selectable, - ) - - def create_lazy_clause(self, reverse_direction: bool = False) -> Tuple[ - ColumnElement[bool], - Dict[str, ColumnElement[Any]], - Dict[ColumnElement[Any], ColumnElement[Any]], - ]: - binds: Dict[ColumnElement[Any], BindParameter[Any]] = {} - equated_columns: Dict[ColumnElement[Any], ColumnElement[Any]] = {} - - has_secondary = self.secondaryjoin is not None - - if has_secondary: - lookup = collections.defaultdict(list) - for l, r in self.local_remote_pairs: - lookup[l].append((l, r)) - equated_columns[r] = l - elif not reverse_direction: - for l, r in self.local_remote_pairs: - equated_columns[r] = l - else: - for l, r in self.local_remote_pairs: - equated_columns[l] = r - - def col_to_bind( - element: ColumnElement[Any], **kw: Any - ) -> Optional[BindParameter[Any]]: - if ( - (not reverse_direction and "local" in element._annotations) - or reverse_direction - and ( - (has_secondary and element in lookup) - or (not has_secondary and "remote" in element._annotations) - ) - ): - if element not in binds: - binds[element] = sql.bindparam( - None, None, type_=element.type, unique=True - ) - return binds[element] - return None - - lazywhere = self.primaryjoin - if self.secondaryjoin is None or not reverse_direction: - lazywhere = visitors.replacement_traverse( - lazywhere, {}, col_to_bind - ) - - if self.secondaryjoin is not None: - secondaryjoin = self.secondaryjoin - if reverse_direction: - secondaryjoin = visitors.replacement_traverse( - secondaryjoin, {}, col_to_bind - ) - lazywhere = sql.and_(lazywhere, secondaryjoin) - - bind_to_col = {binds[col].key: col for col in binds} - - return lazywhere, bind_to_col, equated_columns - - -class _ColInAnnotations: - """Serializable object that tests for names in c._annotations. - - TODO: does this need to be serializable anymore? can we find what the - use case was for that? - - """ - - __slots__ = ("names",) - - def __init__(self, *names: str): - self.names = frozenset(names) - - def __call__(self, c: ClauseElement) -> bool: - return bool(self.names.intersection(c._annotations)) - - -_local_col_exclude = _ColInAnnotations("local", "should_not_adapt") -_remote_col_exclude = _ColInAnnotations("remote", "should_not_adapt") - - -class Relationship( - RelationshipProperty[_T], - _DeclarativeMapped[_T], -): - """Describes an object property that holds a single item or list - of items that correspond to a related database table. - - Public constructor is the :func:`_orm.relationship` function. - - .. seealso:: - - :ref:`relationship_config_toplevel` - - .. versionchanged:: 2.0 Added :class:`_orm.Relationship` as a Declarative - compatible subclass for :class:`_orm.RelationshipProperty`. - - """ - - inherit_cache = True - """:meta private:""" - - -class _RelationshipDeclared( # type: ignore[misc] - Relationship[_T], - WriteOnlyMapped[_T], # not compatible with Mapped[_T] - DynamicMapped[_T], # not compatible with Mapped[_T] -): - """Relationship subclass used implicitly for declarative mapping.""" - - inherit_cache = True - """:meta private:""" - - @classmethod - def _mapper_property_name(cls) -> str: - return "Relationship" |