Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/interfaces.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""
10Contains various base classes used throughout the ORM.
12Defines some key base classes prominent within the internals.
14This module and the classes within are mostly private, though some attributes
15are exposed when inspecting mappings.
17"""
19from __future__ import annotations
21import collections
22import dataclasses
23import typing
24from typing import Any
25from typing import Callable
26from typing import cast
27from typing import ClassVar
28from typing import Dict
29from typing import Generic
30from typing import Iterator
31from typing import List
32from typing import Mapping
33from typing import NamedTuple
34from typing import NoReturn
35from typing import Optional
36from typing import Sequence
37from typing import Set
38from typing import Tuple
39from typing import Type
40from typing import TYPE_CHECKING
41from typing import TypeVar
42from typing import Union
44from . import exc as orm_exc
45from . import path_registry
46from .base import _MappedAttribute as _MappedAttribute
47from .base import EXT_CONTINUE as EXT_CONTINUE # noqa: F401
48from .base import EXT_SKIP as EXT_SKIP # noqa: F401
49from .base import EXT_STOP as EXT_STOP # noqa: F401
50from .base import InspectionAttr as InspectionAttr # noqa: F401
51from .base import InspectionAttrInfo as InspectionAttrInfo
52from .base import MANYTOMANY as MANYTOMANY # noqa: F401
53from .base import MANYTOONE as MANYTOONE # noqa: F401
54from .base import NO_KEY as NO_KEY # noqa: F401
55from .base import NO_VALUE as NO_VALUE # noqa: F401
56from .base import NotExtension as NotExtension # noqa: F401
57from .base import ONETOMANY as ONETOMANY # noqa: F401
58from .base import RelationshipDirection as RelationshipDirection # noqa: F401
59from .base import SQLORMOperations
60from .. import ColumnElement
61from .. import exc as sa_exc
62from .. import inspection
63from .. import util
64from ..sql import operators
65from ..sql import roles
66from ..sql import visitors
67from ..sql.base import _NoArg
68from ..sql.base import ExecutableOption
69from ..sql.cache_key import HasCacheKey
70from ..sql.operators import ColumnOperators
71from ..sql.schema import Column
72from ..sql.type_api import TypeEngine
73from ..util import warn_deprecated
74from ..util.typing import RODescriptorReference
75from ..util.typing import TypedDict
77if typing.TYPE_CHECKING:
78 from ._typing import _EntityType
79 from ._typing import _IdentityKeyType
80 from ._typing import _InstanceDict
81 from ._typing import _InternalEntityType
82 from ._typing import _ORMAdapterProto
83 from .attributes import InstrumentedAttribute
84 from .base import Mapped
85 from .context import _MapperEntity
86 from .context import ORMCompileState
87 from .context import QueryContext
88 from .decl_api import RegistryType
89 from .decl_base import _ClassScanMapperConfig
90 from .loading import _PopulatorDict
91 from .mapper import Mapper
92 from .path_registry import AbstractEntityRegistry
93 from .query import Query
94 from .session import Session
95 from .state import InstanceState
96 from .strategy_options import _LoadElement
97 from .util import AliasedInsp
98 from .util import ORMAdapter
99 from ..engine.result import Result
100 from ..sql._typing import _ColumnExpressionArgument
101 from ..sql._typing import _ColumnsClauseArgument
102 from ..sql._typing import _DMLColumnArgument
103 from ..sql._typing import _InfoType
104 from ..sql.operators import OperatorType
105 from ..sql.visitors import _TraverseInternalsType
106 from ..util.typing import _AnnotationScanType
108_StrategyKey = Tuple[Any, ...]
110_T = TypeVar("_T", bound=Any)
111_T_co = TypeVar("_T_co", bound=Any, covariant=True)
113_TLS = TypeVar("_TLS", bound="Type[LoaderStrategy]")
116class ORMStatementRole(roles.StatementRole):
117 __slots__ = ()
118 _role_name = (
119 "Executable SQL or text() construct, including ORM aware objects"
120 )
123class ORMColumnsClauseRole(
124 roles.ColumnsClauseRole, roles.TypedColumnsClauseRole[_T]
125):
126 __slots__ = ()
127 _role_name = "ORM mapped entity, aliased entity, or Column expression"
130class ORMEntityColumnsClauseRole(ORMColumnsClauseRole[_T]):
131 __slots__ = ()
132 _role_name = "ORM mapped or aliased entity"
135class ORMFromClauseRole(roles.StrictFromClauseRole):
136 __slots__ = ()
137 _role_name = "ORM mapped entity, aliased entity, or FROM expression"
140class ORMColumnDescription(TypedDict):
141 name: str
142 # TODO: add python_type and sql_type here; combining them
143 # into "type" is a bad idea
144 type: Union[Type[Any], TypeEngine[Any]]
145 aliased: bool
146 expr: _ColumnsClauseArgument[Any]
147 entity: Optional[_ColumnsClauseArgument[Any]]
150class _IntrospectsAnnotations:
151 __slots__ = ()
153 @classmethod
154 def _mapper_property_name(cls) -> str:
155 return cls.__name__
157 def found_in_pep593_annotated(self) -> Any:
158 """return a copy of this object to use in declarative when the
159 object is found inside of an Annotated object."""
161 raise NotImplementedError(
162 f"Use of the {self._mapper_property_name()!r} "
163 "construct inside of an Annotated object is not yet supported."
164 )
166 def declarative_scan(
167 self,
168 decl_scan: _ClassScanMapperConfig,
169 registry: RegistryType,
170 cls: Type[Any],
171 originating_module: Optional[str],
172 key: str,
173 mapped_container: Optional[Type[Mapped[Any]]],
174 annotation: Optional[_AnnotationScanType],
175 extracted_mapped_annotation: Optional[_AnnotationScanType],
176 is_dataclass_field: bool,
177 ) -> None:
178 """Perform class-specific initializaton at early declarative scanning
179 time.
181 .. versionadded:: 2.0
183 """
185 def _raise_for_required(self, key: str, cls: Type[Any]) -> NoReturn:
186 raise sa_exc.ArgumentError(
187 f"Python typing annotation is required for attribute "
188 f'"{cls.__name__}.{key}" when primary argument(s) for '
189 f'"{self._mapper_property_name()}" '
190 "construct are None or not present"
191 )
194class _AttributeOptions(NamedTuple):
195 """define Python-local attribute behavior options common to all
196 :class:`.MapperProperty` objects.
198 Currently this includes dataclass-generation arguments.
200 .. versionadded:: 2.0
202 """
204 dataclasses_init: Union[_NoArg, bool]
205 dataclasses_repr: Union[_NoArg, bool]
206 dataclasses_default: Union[_NoArg, Any]
207 dataclasses_default_factory: Union[_NoArg, Callable[[], Any]]
208 dataclasses_compare: Union[_NoArg, bool]
209 dataclasses_kw_only: Union[_NoArg, bool]
210 dataclasses_hash: Union[_NoArg, bool, None]
211 dataclasses_dataclass_metadata: Union[_NoArg, Mapping[Any, Any], None]
213 def _as_dataclass_field(self, key: str) -> Any:
214 """Return a ``dataclasses.Field`` object given these arguments."""
216 kw: Dict[str, Any] = {}
217 if self.dataclasses_default_factory is not _NoArg.NO_ARG:
218 kw["default_factory"] = self.dataclasses_default_factory
219 if self.dataclasses_default is not _NoArg.NO_ARG:
220 kw["default"] = self.dataclasses_default
221 if self.dataclasses_init is not _NoArg.NO_ARG:
222 kw["init"] = self.dataclasses_init
223 if self.dataclasses_repr is not _NoArg.NO_ARG:
224 kw["repr"] = self.dataclasses_repr
225 if self.dataclasses_compare is not _NoArg.NO_ARG:
226 kw["compare"] = self.dataclasses_compare
227 if self.dataclasses_kw_only is not _NoArg.NO_ARG:
228 kw["kw_only"] = self.dataclasses_kw_only
229 if self.dataclasses_hash is not _NoArg.NO_ARG:
230 kw["hash"] = self.dataclasses_hash
231 if self.dataclasses_dataclass_metadata is not _NoArg.NO_ARG:
232 kw["metadata"] = self.dataclasses_dataclass_metadata
234 if "default" in kw and callable(kw["default"]):
235 # callable defaults are ambiguous. deprecate them in favour of
236 # insert_default or default_factory. #9936
237 warn_deprecated(
238 f"Callable object passed to the ``default`` parameter for "
239 f"attribute {key!r} in a ORM-mapped Dataclasses context is "
240 "ambiguous, "
241 "and this use will raise an error in a future release. "
242 "If this callable is intended to produce Core level INSERT "
243 "default values for an underlying ``Column``, use "
244 "the ``mapped_column.insert_default`` parameter instead. "
245 "To establish this callable as providing a default value "
246 "for instances of the dataclass itself, use the "
247 "``default_factory`` dataclasses parameter.",
248 "2.0",
249 )
251 if (
252 "init" in kw
253 and not kw["init"]
254 and "default" in kw
255 and not callable(kw["default"]) # ignore callable defaults. #9936
256 and "default_factory" not in kw # illegal but let dc.field raise
257 ):
258 # fix for #9879
259 default = kw.pop("default")
260 kw["default_factory"] = lambda: default
262 return dataclasses.field(**kw)
264 @classmethod
265 def _get_arguments_for_make_dataclass(
266 cls,
267 key: str,
268 annotation: _AnnotationScanType,
269 mapped_container: Optional[Any],
270 elem: Any,
271 ) -> Union[
272 Tuple[str, _AnnotationScanType],
273 Tuple[str, _AnnotationScanType, dataclasses.Field[Any]],
274 ]:
275 """given attribute key, annotation, and value from a class, return
276 the argument tuple we would pass to dataclasses.make_dataclass()
277 for this attribute.
279 """
280 if isinstance(elem, _DCAttributeOptions):
281 dc_field = elem._attribute_options._as_dataclass_field(key)
283 return (key, annotation, dc_field)
284 elif elem is not _NoArg.NO_ARG:
285 # why is typing not erroring on this?
286 return (key, annotation, elem)
287 elif mapped_container is not None:
288 # it's Mapped[], but there's no "element", which means declarative
289 # did not actually do anything for this field. this shouldn't
290 # happen.
291 # previously, this would occur because _scan_attributes would
292 # skip a field that's on an already mapped superclass, but it
293 # would still include it in the annotations, leading
294 # to issue #8718
296 assert False, "Mapped[] received without a mapping declaration"
298 else:
299 # plain dataclass field, not mapped. Is only possible
300 # if __allow_unmapped__ is set up. I can see this mode causing
301 # problems...
302 return (key, annotation)
305_DEFAULT_ATTRIBUTE_OPTIONS = _AttributeOptions(
306 _NoArg.NO_ARG,
307 _NoArg.NO_ARG,
308 _NoArg.NO_ARG,
309 _NoArg.NO_ARG,
310 _NoArg.NO_ARG,
311 _NoArg.NO_ARG,
312 _NoArg.NO_ARG,
313 _NoArg.NO_ARG,
314)
316_DEFAULT_READONLY_ATTRIBUTE_OPTIONS = _AttributeOptions(
317 False,
318 _NoArg.NO_ARG,
319 _NoArg.NO_ARG,
320 _NoArg.NO_ARG,
321 _NoArg.NO_ARG,
322 _NoArg.NO_ARG,
323 _NoArg.NO_ARG,
324 _NoArg.NO_ARG,
325)
328class _DCAttributeOptions:
329 """mixin for descriptors or configurational objects that include dataclass
330 field options.
332 This includes :class:`.MapperProperty`, :class:`._MapsColumn` within
333 the ORM, but also includes :class:`.AssociationProxy` within ext.
334 Can in theory be used for other descriptors that serve a similar role
335 as association proxy. (*maybe* hybrids, not sure yet.)
337 """
339 __slots__ = ()
341 _attribute_options: _AttributeOptions
342 """behavioral options for ORM-enabled Python attributes
344 .. versionadded:: 2.0
346 """
348 _has_dataclass_arguments: bool
351class _MapsColumns(_DCAttributeOptions, _MappedAttribute[_T]):
352 """interface for declarative-capable construct that delivers one or more
353 Column objects to the declarative process to be part of a Table.
354 """
356 __slots__ = ()
358 @property
359 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
360 """return a MapperProperty to be assigned to the declarative mapping"""
361 raise NotImplementedError()
363 @property
364 def columns_to_assign(self) -> List[Tuple[Column[_T], int]]:
365 """A list of Column objects that should be declaratively added to the
366 new Table object.
368 """
369 raise NotImplementedError()
372# NOTE: MapperProperty needs to extend _MappedAttribute so that declarative
373# typing works, i.e. "Mapped[A] = relationship()". This introduces an
374# inconvenience which is that all the MapperProperty objects are treated
375# as descriptors by typing tools, which are misled by this as assignment /
376# access to a descriptor attribute wants to move through __get__.
377# Therefore, references to MapperProperty as an instance variable, such
378# as in PropComparator, may have some special typing workarounds such as the
379# use of sqlalchemy.util.typing.DescriptorReference to avoid mis-interpretation
380# by typing tools
381@inspection._self_inspects
382class MapperProperty(
383 HasCacheKey,
384 _DCAttributeOptions,
385 _MappedAttribute[_T],
386 InspectionAttrInfo,
387 util.MemoizedSlots,
388):
389 """Represent a particular class attribute mapped by :class:`_orm.Mapper`.
391 The most common occurrences of :class:`.MapperProperty` are the
392 mapped :class:`_schema.Column`, which is represented in a mapping as
393 an instance of :class:`.ColumnProperty`,
394 and a reference to another class produced by :func:`_orm.relationship`,
395 represented in the mapping as an instance of
396 :class:`.Relationship`.
398 """
400 __slots__ = (
401 "_configure_started",
402 "_configure_finished",
403 "_attribute_options",
404 "_has_dataclass_arguments",
405 "parent",
406 "key",
407 "info",
408 "doc",
409 )
411 _cache_key_traversal: _TraverseInternalsType = [
412 ("parent", visitors.ExtendedInternalTraversal.dp_has_cache_key),
413 ("key", visitors.ExtendedInternalTraversal.dp_string),
414 ]
416 if not TYPE_CHECKING:
417 cascade = None
419 is_property = True
420 """Part of the InspectionAttr interface; states this object is a
421 mapper property.
423 """
425 comparator: PropComparator[_T]
426 """The :class:`_orm.PropComparator` instance that implements SQL
427 expression construction on behalf of this mapped attribute."""
429 key: str
430 """name of class attribute"""
432 parent: Mapper[Any]
433 """the :class:`.Mapper` managing this property."""
435 _is_relationship = False
437 _links_to_entity: bool
438 """True if this MapperProperty refers to a mapped entity.
440 Should only be True for Relationship, False for all others.
442 """
444 doc: Optional[str]
445 """optional documentation string"""
447 info: _InfoType
448 """Info dictionary associated with the object, allowing user-defined
449 data to be associated with this :class:`.InspectionAttr`.
451 The dictionary is generated when first accessed. Alternatively,
452 it can be specified as a constructor argument to the
453 :func:`.column_property`, :func:`_orm.relationship`, or :func:`.composite`
454 functions.
456 .. seealso::
458 :attr:`.QueryableAttribute.info`
460 :attr:`.SchemaItem.info`
462 """
464 def _memoized_attr_info(self) -> _InfoType:
465 """Info dictionary associated with the object, allowing user-defined
466 data to be associated with this :class:`.InspectionAttr`.
468 The dictionary is generated when first accessed. Alternatively,
469 it can be specified as a constructor argument to the
470 :func:`.column_property`, :func:`_orm.relationship`, or
471 :func:`.composite`
472 functions.
474 .. seealso::
476 :attr:`.QueryableAttribute.info`
478 :attr:`.SchemaItem.info`
480 """
481 return {}
483 def setup(
484 self,
485 context: ORMCompileState,
486 query_entity: _MapperEntity,
487 path: AbstractEntityRegistry,
488 adapter: Optional[ORMAdapter],
489 **kwargs: Any,
490 ) -> None:
491 """Called by Query for the purposes of constructing a SQL statement.
493 Each MapperProperty associated with the target mapper processes the
494 statement referenced by the query context, adding columns and/or
495 criterion as appropriate.
497 """
499 def create_row_processor(
500 self,
501 context: ORMCompileState,
502 query_entity: _MapperEntity,
503 path: AbstractEntityRegistry,
504 mapper: Mapper[Any],
505 result: Result[Any],
506 adapter: Optional[ORMAdapter],
507 populators: _PopulatorDict,
508 ) -> None:
509 """Produce row processing functions and append to the given
510 set of populators lists.
512 """
514 def cascade_iterator(
515 self,
516 type_: str,
517 state: InstanceState[Any],
518 dict_: _InstanceDict,
519 visited_states: Set[InstanceState[Any]],
520 halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
521 ) -> Iterator[
522 Tuple[object, Mapper[Any], InstanceState[Any], _InstanceDict]
523 ]:
524 """Iterate through instances related to the given instance for
525 a particular 'cascade', starting with this MapperProperty.
527 Return an iterator3-tuples (instance, mapper, state).
529 Note that the 'cascade' collection on this MapperProperty is
530 checked first for the given type before cascade_iterator is called.
532 This method typically only applies to Relationship.
534 """
536 return iter(())
538 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
539 """Set the parent mapper that references this MapperProperty.
541 This method is overridden by some subclasses to perform extra
542 setup when the mapper is first known.
544 """
545 self.parent = parent
547 def instrument_class(self, mapper: Mapper[Any]) -> None:
548 """Hook called by the Mapper to the property to initiate
549 instrumentation of the class attribute managed by this
550 MapperProperty.
552 The MapperProperty here will typically call out to the
553 attributes module to set up an InstrumentedAttribute.
555 This step is the first of two steps to set up an InstrumentedAttribute,
556 and is called early in the mapper setup process.
558 The second step is typically the init_class_attribute step,
559 called from StrategizedProperty via the post_instrument_class()
560 hook. This step assigns additional state to the InstrumentedAttribute
561 (specifically the "impl") which has been determined after the
562 MapperProperty has determined what kind of persistence
563 management it needs to do (e.g. scalar, object, collection, etc).
565 """
567 def __init__(
568 self,
569 attribute_options: Optional[_AttributeOptions] = None,
570 _assume_readonly_dc_attributes: bool = False,
571 ) -> None:
572 self._configure_started = False
573 self._configure_finished = False
575 if _assume_readonly_dc_attributes:
576 default_attrs = _DEFAULT_READONLY_ATTRIBUTE_OPTIONS
577 else:
578 default_attrs = _DEFAULT_ATTRIBUTE_OPTIONS
580 if attribute_options and attribute_options != default_attrs:
581 self._has_dataclass_arguments = True
582 self._attribute_options = attribute_options
583 else:
584 self._has_dataclass_arguments = False
585 self._attribute_options = default_attrs
587 def init(self) -> None:
588 """Called after all mappers are created to assemble
589 relationships between mappers and perform other post-mapper-creation
590 initialization steps.
593 """
594 self._configure_started = True
595 self.do_init()
596 self._configure_finished = True
598 @property
599 def class_attribute(self) -> InstrumentedAttribute[_T]:
600 """Return the class-bound descriptor corresponding to this
601 :class:`.MapperProperty`.
603 This is basically a ``getattr()`` call::
605 return getattr(self.parent.class_, self.key)
607 I.e. if this :class:`.MapperProperty` were named ``addresses``,
608 and the class to which it is mapped is ``User``, this sequence
609 is possible::
611 >>> from sqlalchemy import inspect
612 >>> mapper = inspect(User)
613 >>> addresses_property = mapper.attrs.addresses
614 >>> addresses_property.class_attribute is User.addresses
615 True
616 >>> User.addresses.property is addresses_property
617 True
620 """
622 return getattr(self.parent.class_, self.key) # type: ignore
624 def do_init(self) -> None:
625 """Perform subclass-specific initialization post-mapper-creation
626 steps.
628 This is a template method called by the ``MapperProperty``
629 object's init() method.
631 """
633 def post_instrument_class(self, mapper: Mapper[Any]) -> None:
634 """Perform instrumentation adjustments that need to occur
635 after init() has completed.
637 The given Mapper is the Mapper invoking the operation, which
638 may not be the same Mapper as self.parent in an inheritance
639 scenario; however, Mapper will always at least be a sub-mapper of
640 self.parent.
642 This method is typically used by StrategizedProperty, which delegates
643 it to LoaderStrategy.init_class_attribute() to perform final setup
644 on the class-bound InstrumentedAttribute.
646 """
648 def merge(
649 self,
650 session: Session,
651 source_state: InstanceState[Any],
652 source_dict: _InstanceDict,
653 dest_state: InstanceState[Any],
654 dest_dict: _InstanceDict,
655 load: bool,
656 _recursive: Dict[Any, object],
657 _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
658 ) -> None:
659 """Merge the attribute represented by this ``MapperProperty``
660 from source to destination object.
662 """
664 def __repr__(self) -> str:
665 return "<%s at 0x%x; %s>" % (
666 self.__class__.__name__,
667 id(self),
668 getattr(self, "key", "no key"),
669 )
672@inspection._self_inspects
673class PropComparator(SQLORMOperations[_T_co], Generic[_T_co], ColumnOperators):
674 r"""Defines SQL operations for ORM mapped attributes.
676 SQLAlchemy allows for operators to
677 be redefined at both the Core and ORM level. :class:`.PropComparator`
678 is the base class of operator redefinition for ORM-level operations,
679 including those of :class:`.ColumnProperty`,
680 :class:`.Relationship`, and :class:`.Composite`.
682 User-defined subclasses of :class:`.PropComparator` may be created. The
683 built-in Python comparison and math operator methods, such as
684 :meth:`.operators.ColumnOperators.__eq__`,
685 :meth:`.operators.ColumnOperators.__lt__`, and
686 :meth:`.operators.ColumnOperators.__add__`, can be overridden to provide
687 new operator behavior. The custom :class:`.PropComparator` is passed to
688 the :class:`.MapperProperty` instance via the ``comparator_factory``
689 argument. In each case,
690 the appropriate subclass of :class:`.PropComparator` should be used::
692 # definition of custom PropComparator subclasses
694 from sqlalchemy.orm.properties import (
695 ColumnProperty,
696 Composite,
697 Relationship,
698 )
701 class MyColumnComparator(ColumnProperty.Comparator):
702 def __eq__(self, other):
703 return self.__clause_element__() == other
706 class MyRelationshipComparator(Relationship.Comparator):
707 def any(self, expression):
708 "define the 'any' operation"
709 # ...
712 class MyCompositeComparator(Composite.Comparator):
713 def __gt__(self, other):
714 "redefine the 'greater than' operation"
716 return sql.and_(
717 *[
718 a > b
719 for a, b in zip(
720 self.__clause_element__().clauses,
721 other.__composite_values__(),
722 )
723 ]
724 )
727 # application of custom PropComparator subclasses
729 from sqlalchemy.orm import column_property, relationship, composite
730 from sqlalchemy import Column, String
733 class SomeMappedClass(Base):
734 some_column = column_property(
735 Column("some_column", String),
736 comparator_factory=MyColumnComparator,
737 )
739 some_relationship = relationship(
740 SomeOtherClass, comparator_factory=MyRelationshipComparator
741 )
743 some_composite = composite(
744 Column("a", String),
745 Column("b", String),
746 comparator_factory=MyCompositeComparator,
747 )
749 Note that for column-level operator redefinition, it's usually
750 simpler to define the operators at the Core level, using the
751 :attr:`.TypeEngine.comparator_factory` attribute. See
752 :ref:`types_operators` for more detail.
754 .. seealso::
756 :class:`.ColumnProperty.Comparator`
758 :class:`.Relationship.Comparator`
760 :class:`.Composite.Comparator`
762 :class:`.ColumnOperators`
764 :ref:`types_operators`
766 :attr:`.TypeEngine.comparator_factory`
768 """
770 __slots__ = "prop", "_parententity", "_adapt_to_entity"
772 __visit_name__ = "orm_prop_comparator"
774 _parententity: _InternalEntityType[Any]
775 _adapt_to_entity: Optional[AliasedInsp[Any]]
776 prop: RODescriptorReference[MapperProperty[_T_co]]
778 def __init__(
779 self,
780 prop: MapperProperty[_T],
781 parentmapper: _InternalEntityType[Any],
782 adapt_to_entity: Optional[AliasedInsp[Any]] = None,
783 ):
784 self.prop = prop
785 self._parententity = adapt_to_entity or parentmapper
786 self._adapt_to_entity = adapt_to_entity
788 @util.non_memoized_property
789 def property(self) -> MapperProperty[_T_co]:
790 """Return the :class:`.MapperProperty` associated with this
791 :class:`.PropComparator`.
794 Return values here will commonly be instances of
795 :class:`.ColumnProperty` or :class:`.Relationship`.
798 """
799 return self.prop
801 def __clause_element__(self) -> roles.ColumnsClauseRole:
802 raise NotImplementedError("%r" % self)
804 def _bulk_update_tuples(
805 self, value: Any
806 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
807 """Receive a SQL expression that represents a value in the SET
808 clause of an UPDATE statement.
810 Return a tuple that can be passed to a :class:`_expression.Update`
811 construct.
813 """
815 return [(cast("_DMLColumnArgument", self.__clause_element__()), value)]
817 def adapt_to_entity(
818 self, adapt_to_entity: AliasedInsp[Any]
819 ) -> PropComparator[_T_co]:
820 """Return a copy of this PropComparator which will use the given
821 :class:`.AliasedInsp` to produce corresponding expressions.
822 """
823 return self.__class__(self.prop, self._parententity, adapt_to_entity)
825 @util.ro_non_memoized_property
826 def _parentmapper(self) -> Mapper[Any]:
827 """legacy; this is renamed to _parententity to be
828 compatible with QueryableAttribute."""
829 return self._parententity.mapper
831 def _criterion_exists(
832 self,
833 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
834 **kwargs: Any,
835 ) -> ColumnElement[Any]:
836 return self.prop.comparator._criterion_exists(criterion, **kwargs)
838 @util.ro_non_memoized_property
839 def adapter(self) -> Optional[_ORMAdapterProto]:
840 """Produce a callable that adapts column expressions
841 to suit an aliased version of this comparator.
843 """
844 if self._adapt_to_entity is None:
845 return None
846 else:
847 return self._adapt_to_entity._orm_adapt_element
849 @util.ro_non_memoized_property
850 def info(self) -> _InfoType:
851 return self.prop.info
853 @staticmethod
854 def _any_op(a: Any, b: Any, **kwargs: Any) -> Any:
855 return a.any(b, **kwargs)
857 @staticmethod
858 def _has_op(left: Any, other: Any, **kwargs: Any) -> Any:
859 return left.has(other, **kwargs)
861 @staticmethod
862 def _of_type_op(a: Any, class_: Any) -> Any:
863 return a.of_type(class_)
865 any_op = cast(operators.OperatorType, _any_op)
866 has_op = cast(operators.OperatorType, _has_op)
867 of_type_op = cast(operators.OperatorType, _of_type_op)
869 if typing.TYPE_CHECKING:
871 def operate(
872 self, op: OperatorType, *other: Any, **kwargs: Any
873 ) -> ColumnElement[Any]: ...
875 def reverse_operate(
876 self, op: OperatorType, other: Any, **kwargs: Any
877 ) -> ColumnElement[Any]: ...
879 def of_type(self, class_: _EntityType[Any]) -> PropComparator[_T_co]:
880 r"""Redefine this object in terms of a polymorphic subclass,
881 :func:`_orm.with_polymorphic` construct, or :func:`_orm.aliased`
882 construct.
884 Returns a new PropComparator from which further criterion can be
885 evaluated.
887 e.g.::
889 query.join(Company.employees.of_type(Engineer)).filter(
890 Engineer.name == "foo"
891 )
893 :param \class_: a class or mapper indicating that criterion will be
894 against this specific subclass.
896 .. seealso::
898 :ref:`orm_queryguide_joining_relationships_aliased` - in the
899 :ref:`queryguide_toplevel`
901 :ref:`inheritance_of_type`
903 """
905 return self.operate(PropComparator.of_type_op, class_) # type: ignore
907 def and_(
908 self, *criteria: _ColumnExpressionArgument[bool]
909 ) -> PropComparator[bool]:
910 """Add additional criteria to the ON clause that's represented by this
911 relationship attribute.
913 E.g.::
916 stmt = select(User).join(
917 User.addresses.and_(Address.email_address != "foo")
918 )
920 stmt = select(User).options(
921 joinedload(User.addresses.and_(Address.email_address != "foo"))
922 )
924 .. versionadded:: 1.4
926 .. seealso::
928 :ref:`orm_queryguide_join_on_augmented`
930 :ref:`loader_option_criteria`
932 :func:`.with_loader_criteria`
934 """
935 return self.operate(operators.and_, *criteria) # type: ignore
937 def any(
938 self,
939 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
940 **kwargs: Any,
941 ) -> ColumnElement[bool]:
942 r"""Return a SQL expression representing true if this element
943 references a member which meets the given criterion.
945 The usual implementation of ``any()`` is
946 :meth:`.Relationship.Comparator.any`.
948 :param criterion: an optional ClauseElement formulated against the
949 member class' table or attributes.
951 :param \**kwargs: key/value pairs corresponding to member class
952 attribute names which will be compared via equality to the
953 corresponding values.
955 """
957 return self.operate(PropComparator.any_op, criterion, **kwargs)
959 def has(
960 self,
961 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
962 **kwargs: Any,
963 ) -> ColumnElement[bool]:
964 r"""Return a SQL expression representing true if this element
965 references a member which meets the given criterion.
967 The usual implementation of ``has()`` is
968 :meth:`.Relationship.Comparator.has`.
970 :param criterion: an optional ClauseElement formulated against the
971 member class' table or attributes.
973 :param \**kwargs: key/value pairs corresponding to member class
974 attribute names which will be compared via equality to the
975 corresponding values.
977 """
979 return self.operate(PropComparator.has_op, criterion, **kwargs)
982class StrategizedProperty(MapperProperty[_T]):
983 """A MapperProperty which uses selectable strategies to affect
984 loading behavior.
986 There is a single strategy selected by default. Alternate
987 strategies can be selected at Query time through the usage of
988 ``StrategizedOption`` objects via the Query.options() method.
990 The mechanics of StrategizedProperty are used for every Query
991 invocation for every mapped attribute participating in that Query,
992 to determine first how the attribute will be rendered in SQL
993 and secondly how the attribute will retrieve a value from a result
994 row and apply it to a mapped object. The routines here are very
995 performance-critical.
997 """
999 __slots__ = (
1000 "_strategies",
1001 "strategy",
1002 "_wildcard_token",
1003 "_default_path_loader_key",
1004 "strategy_key",
1005 )
1006 inherit_cache = True
1007 strategy_wildcard_key: ClassVar[str]
1009 strategy_key: _StrategyKey
1011 _strategies: Dict[_StrategyKey, LoaderStrategy]
1013 def _memoized_attr__wildcard_token(self) -> Tuple[str]:
1014 return (
1015 f"{self.strategy_wildcard_key}:{path_registry._WILDCARD_TOKEN}",
1016 )
1018 def _memoized_attr__default_path_loader_key(
1019 self,
1020 ) -> Tuple[str, Tuple[str]]:
1021 return (
1022 "loader",
1023 (f"{self.strategy_wildcard_key}:{path_registry._DEFAULT_TOKEN}",),
1024 )
1026 def _get_context_loader(
1027 self, context: ORMCompileState, path: AbstractEntityRegistry
1028 ) -> Optional[_LoadElement]:
1029 load: Optional[_LoadElement] = None
1031 search_path = path[self]
1033 # search among: exact match, "attr.*", "default" strategy
1034 # if any.
1035 for path_key in (
1036 search_path._loader_key,
1037 search_path._wildcard_path_loader_key,
1038 search_path._default_path_loader_key,
1039 ):
1040 if path_key in context.attributes:
1041 load = context.attributes[path_key]
1042 break
1044 # note that if strategy_options.Load is placing non-actionable
1045 # objects in the context like defaultload(), we would
1046 # need to continue the loop here if we got such an
1047 # option as below.
1048 # if load.strategy or load.local_opts:
1049 # break
1051 return load
1053 def _get_strategy(self, key: _StrategyKey) -> LoaderStrategy:
1054 try:
1055 return self._strategies[key]
1056 except KeyError:
1057 pass
1059 # run outside to prevent transfer of exception context
1060 cls = self._strategy_lookup(self, *key)
1061 # this previously was setting self._strategies[cls], that's
1062 # a bad idea; should use strategy key at all times because every
1063 # strategy has multiple keys at this point
1064 self._strategies[key] = strategy = cls(self, key)
1065 return strategy
1067 def setup(
1068 self,
1069 context: ORMCompileState,
1070 query_entity: _MapperEntity,
1071 path: AbstractEntityRegistry,
1072 adapter: Optional[ORMAdapter],
1073 **kwargs: Any,
1074 ) -> None:
1075 loader = self._get_context_loader(context, path)
1076 if loader and loader.strategy:
1077 strat = self._get_strategy(loader.strategy)
1078 else:
1079 strat = self.strategy
1080 strat.setup_query(
1081 context, query_entity, path, loader, adapter, **kwargs
1082 )
1084 def create_row_processor(
1085 self,
1086 context: ORMCompileState,
1087 query_entity: _MapperEntity,
1088 path: AbstractEntityRegistry,
1089 mapper: Mapper[Any],
1090 result: Result[Any],
1091 adapter: Optional[ORMAdapter],
1092 populators: _PopulatorDict,
1093 ) -> None:
1094 loader = self._get_context_loader(context, path)
1095 if loader and loader.strategy:
1096 strat = self._get_strategy(loader.strategy)
1097 else:
1098 strat = self.strategy
1099 strat.create_row_processor(
1100 context,
1101 query_entity,
1102 path,
1103 loader,
1104 mapper,
1105 result,
1106 adapter,
1107 populators,
1108 )
1110 def do_init(self) -> None:
1111 self._strategies = {}
1112 self.strategy = self._get_strategy(self.strategy_key)
1114 def post_instrument_class(self, mapper: Mapper[Any]) -> None:
1115 if (
1116 not self.parent.non_primary
1117 and not mapper.class_manager._attr_has_impl(self.key)
1118 ):
1119 self.strategy.init_class_attribute(mapper)
1121 _all_strategies: collections.defaultdict[
1122 Type[MapperProperty[Any]], Dict[_StrategyKey, Type[LoaderStrategy]]
1123 ] = collections.defaultdict(dict)
1125 @classmethod
1126 def strategy_for(cls, **kw: Any) -> Callable[[_TLS], _TLS]:
1127 def decorate(dec_cls: _TLS) -> _TLS:
1128 # ensure each subclass of the strategy has its
1129 # own _strategy_keys collection
1130 if "_strategy_keys" not in dec_cls.__dict__:
1131 dec_cls._strategy_keys = []
1132 key = tuple(sorted(kw.items()))
1133 cls._all_strategies[cls][key] = dec_cls
1134 dec_cls._strategy_keys.append(key)
1135 return dec_cls
1137 return decorate
1139 @classmethod
1140 def _strategy_lookup(
1141 cls, requesting_property: MapperProperty[Any], *key: Any
1142 ) -> Type[LoaderStrategy]:
1143 requesting_property.parent._with_polymorphic_mappers
1145 for prop_cls in cls.__mro__:
1146 if prop_cls in cls._all_strategies:
1147 if TYPE_CHECKING:
1148 assert issubclass(prop_cls, MapperProperty)
1149 strategies = cls._all_strategies[prop_cls]
1150 try:
1151 return strategies[key]
1152 except KeyError:
1153 pass
1155 for property_type, strats in cls._all_strategies.items():
1156 if key in strats:
1157 intended_property_type = property_type
1158 actual_strategy = strats[key]
1159 break
1160 else:
1161 intended_property_type = None
1162 actual_strategy = None
1164 raise orm_exc.LoaderStrategyException(
1165 cls,
1166 requesting_property,
1167 intended_property_type,
1168 actual_strategy,
1169 key,
1170 )
1173class ORMOption(ExecutableOption):
1174 """Base class for option objects that are passed to ORM queries.
1176 These options may be consumed by :meth:`.Query.options`,
1177 :meth:`.Select.options`, or in a more general sense by any
1178 :meth:`.Executable.options` method. They are interpreted at
1179 statement compile time or execution time in modern use. The
1180 deprecated :class:`.MapperOption` is consumed at ORM query construction
1181 time.
1183 .. versionadded:: 1.4
1185 """
1187 __slots__ = ()
1189 _is_legacy_option = False
1191 propagate_to_loaders = False
1192 """if True, indicate this option should be carried along
1193 to "secondary" SELECT statements that occur for relationship
1194 lazy loaders as well as attribute load / refresh operations.
1196 """
1198 _is_core = False
1200 _is_user_defined = False
1202 _is_compile_state = False
1204 _is_criteria_option = False
1206 _is_strategy_option = False
1208 def _adapt_cached_option_to_uncached_option(
1209 self, context: QueryContext, uncached_opt: ORMOption
1210 ) -> ORMOption:
1211 """adapt this option to the "uncached" version of itself in a
1212 loader strategy context.
1214 given "self" which is an option from a cached query, as well as the
1215 corresponding option from the uncached version of the same query,
1216 return the option we should use in a new query, in the context of a
1217 loader strategy being asked to load related rows on behalf of that
1218 cached query, which is assumed to be building a new query based on
1219 entities passed to us from the cached query.
1221 Currently this routine chooses between "self" and "uncached" without
1222 manufacturing anything new. If the option is itself a loader strategy
1223 option which has a path, that path needs to match to the entities being
1224 passed to us by the cached query, so the :class:`_orm.Load` subclass
1225 overrides this to return "self". For all other options, we return the
1226 uncached form which may have changing state, such as a
1227 with_loader_criteria() option which will very often have new state.
1229 This routine could in the future involve
1230 generating a new option based on both inputs if use cases arise,
1231 such as if with_loader_criteria() needed to match up to
1232 ``AliasedClass`` instances given in the parent query.
1234 However, longer term it might be better to restructure things such that
1235 ``AliasedClass`` entities are always matched up on their cache key,
1236 instead of identity, in things like paths and such, so that this whole
1237 issue of "the uncached option does not match the entities" goes away.
1238 However this would make ``PathRegistry`` more complicated and difficult
1239 to debug as well as potentially less performant in that it would be
1240 hashing enormous cache keys rather than a simple AliasedInsp. UNLESS,
1241 we could get cache keys overall to be reliably hashed into something
1242 like an md5 key.
1244 .. versionadded:: 1.4.41
1246 """
1247 if uncached_opt is not None:
1248 return uncached_opt
1249 else:
1250 return self
1253class CompileStateOption(HasCacheKey, ORMOption):
1254 """base for :class:`.ORMOption` classes that affect the compilation of
1255 a SQL query and therefore need to be part of the cache key.
1257 .. note:: :class:`.CompileStateOption` is generally non-public and
1258 should not be used as a base class for user-defined options; instead,
1259 use :class:`.UserDefinedOption`, which is easier to use as it does not
1260 interact with ORM compilation internals or caching.
1262 :class:`.CompileStateOption` defines an internal attribute
1263 ``_is_compile_state=True`` which has the effect of the ORM compilation
1264 routines for SELECT and other statements will call upon these options when
1265 a SQL string is being compiled. As such, these classes implement
1266 :class:`.HasCacheKey` and need to provide robust ``_cache_key_traversal``
1267 structures.
1269 The :class:`.CompileStateOption` class is used to implement the ORM
1270 :class:`.LoaderOption` and :class:`.CriteriaOption` classes.
1272 .. versionadded:: 1.4.28
1275 """
1277 __slots__ = ()
1279 _is_compile_state = True
1281 def process_compile_state(self, compile_state: ORMCompileState) -> None:
1282 """Apply a modification to a given :class:`.ORMCompileState`.
1284 This method is part of the implementation of a particular
1285 :class:`.CompileStateOption` and is only invoked internally
1286 when an ORM query is compiled.
1288 """
1290 def process_compile_state_replaced_entities(
1291 self,
1292 compile_state: ORMCompileState,
1293 mapper_entities: Sequence[_MapperEntity],
1294 ) -> None:
1295 """Apply a modification to a given :class:`.ORMCompileState`,
1296 given entities that were replaced by with_only_columns() or
1297 with_entities().
1299 This method is part of the implementation of a particular
1300 :class:`.CompileStateOption` and is only invoked internally
1301 when an ORM query is compiled.
1303 .. versionadded:: 1.4.19
1305 """
1308class LoaderOption(CompileStateOption):
1309 """Describe a loader modification to an ORM statement at compilation time.
1311 .. versionadded:: 1.4
1313 """
1315 __slots__ = ()
1317 def process_compile_state_replaced_entities(
1318 self,
1319 compile_state: ORMCompileState,
1320 mapper_entities: Sequence[_MapperEntity],
1321 ) -> None:
1322 self.process_compile_state(compile_state)
1325class CriteriaOption(CompileStateOption):
1326 """Describe a WHERE criteria modification to an ORM statement at
1327 compilation time.
1329 .. versionadded:: 1.4
1331 """
1333 __slots__ = ()
1335 _is_criteria_option = True
1337 def get_global_criteria(self, attributes: Dict[str, Any]) -> None:
1338 """update additional entity criteria options in the given
1339 attributes dictionary.
1341 """
1344class UserDefinedOption(ORMOption):
1345 """Base class for a user-defined option that can be consumed from the
1346 :meth:`.SessionEvents.do_orm_execute` event hook.
1348 """
1350 __slots__ = ("payload",)
1352 _is_legacy_option = False
1354 _is_user_defined = True
1356 propagate_to_loaders = False
1357 """if True, indicate this option should be carried along
1358 to "secondary" Query objects produced during lazy loads
1359 or refresh operations.
1361 """
1363 def __init__(self, payload: Optional[Any] = None):
1364 self.payload = payload
1367@util.deprecated_cls(
1368 "1.4",
1369 "The :class:`.MapperOption class is deprecated and will be removed "
1370 "in a future release. For "
1371 "modifications to queries on a per-execution basis, use the "
1372 ":class:`.UserDefinedOption` class to establish state within a "
1373 ":class:`.Query` or other Core statement, then use the "
1374 ":meth:`.SessionEvents.before_orm_execute` hook to consume them.",
1375 constructor=None,
1376)
1377class MapperOption(ORMOption):
1378 """Describe a modification to a Query"""
1380 __slots__ = ()
1382 _is_legacy_option = True
1384 propagate_to_loaders = False
1385 """if True, indicate this option should be carried along
1386 to "secondary" Query objects produced during lazy loads
1387 or refresh operations.
1389 """
1391 def process_query(self, query: Query[Any]) -> None:
1392 """Apply a modification to the given :class:`_query.Query`."""
1394 def process_query_conditionally(self, query: Query[Any]) -> None:
1395 """same as process_query(), except that this option may not
1396 apply to the given query.
1398 This is typically applied during a lazy load or scalar refresh
1399 operation to propagate options stated in the original Query to the
1400 new Query being used for the load. It occurs for those options that
1401 specify propagate_to_loaders=True.
1403 """
1405 self.process_query(query)
1408class LoaderStrategy:
1409 """Describe the loading behavior of a StrategizedProperty object.
1411 The ``LoaderStrategy`` interacts with the querying process in three
1412 ways:
1414 * it controls the configuration of the ``InstrumentedAttribute``
1415 placed on a class to handle the behavior of the attribute. this
1416 may involve setting up class-level callable functions to fire
1417 off a select operation when the attribute is first accessed
1418 (i.e. a lazy load)
1420 * it processes the ``QueryContext`` at statement construction time,
1421 where it can modify the SQL statement that is being produced.
1422 For example, simple column attributes will add their represented
1423 column to the list of selected columns, a joined eager loader
1424 may establish join clauses to add to the statement.
1426 * It produces "row processor" functions at result fetching time.
1427 These "row processor" functions populate a particular attribute
1428 on a particular mapped instance.
1430 """
1432 __slots__ = (
1433 "parent_property",
1434 "is_class_level",
1435 "parent",
1436 "key",
1437 "strategy_key",
1438 "strategy_opts",
1439 )
1441 _strategy_keys: ClassVar[List[_StrategyKey]]
1443 def __init__(
1444 self, parent: MapperProperty[Any], strategy_key: _StrategyKey
1445 ):
1446 self.parent_property = parent
1447 self.is_class_level = False
1448 self.parent = self.parent_property.parent
1449 self.key = self.parent_property.key
1450 self.strategy_key = strategy_key
1451 self.strategy_opts = dict(strategy_key)
1453 def init_class_attribute(self, mapper: Mapper[Any]) -> None:
1454 pass
1456 def setup_query(
1457 self,
1458 compile_state: ORMCompileState,
1459 query_entity: _MapperEntity,
1460 path: AbstractEntityRegistry,
1461 loadopt: Optional[_LoadElement],
1462 adapter: Optional[ORMAdapter],
1463 **kwargs: Any,
1464 ) -> None:
1465 """Establish column and other state for a given QueryContext.
1467 This method fulfills the contract specified by MapperProperty.setup().
1469 StrategizedProperty delegates its setup() method
1470 directly to this method.
1472 """
1474 def create_row_processor(
1475 self,
1476 context: ORMCompileState,
1477 query_entity: _MapperEntity,
1478 path: AbstractEntityRegistry,
1479 loadopt: Optional[_LoadElement],
1480 mapper: Mapper[Any],
1481 result: Result[Any],
1482 adapter: Optional[ORMAdapter],
1483 populators: _PopulatorDict,
1484 ) -> None:
1485 """Establish row processing functions for a given QueryContext.
1487 This method fulfills the contract specified by
1488 MapperProperty.create_row_processor().
1490 StrategizedProperty delegates its create_row_processor() method
1491 directly to this method.
1493 """
1495 def __str__(self) -> str:
1496 return str(self.parent_property)