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 NamedTuple
33from typing import NoReturn
34from typing import Optional
35from typing import Sequence
36from typing import Set
37from typing import Tuple
38from typing import Type
39from typing import TYPE_CHECKING
40from typing import TypedDict
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 DONT_SET as DONT_SET # noqa: F401
48from .base import EXT_CONTINUE as EXT_CONTINUE # noqa: F401
49from .base import EXT_SKIP as EXT_SKIP # noqa: F401
50from .base import EXT_STOP as EXT_STOP # noqa: F401
51from .base import InspectionAttr as InspectionAttr # noqa: F401
52from .base import InspectionAttrInfo as InspectionAttrInfo
53from .base import MANYTOMANY as MANYTOMANY # noqa: F401
54from .base import MANYTOONE as MANYTOONE # noqa: F401
55from .base import NO_KEY as NO_KEY # noqa: F401
56from .base import NO_VALUE as NO_VALUE # noqa: F401
57from .base import NotExtension as NotExtension # noqa: F401
58from .base import ONETOMANY as ONETOMANY # noqa: F401
59from .base import RelationshipDirection as RelationshipDirection # noqa: F401
60from .base import SQLORMOperations
61from .. import ColumnElement
62from .. import exc as sa_exc
63from .. import inspection
64from .. import util
65from ..sql import operators
66from ..sql import roles
67from ..sql import visitors
68from ..sql.base import _NoArg
69from ..sql.base import ExecutableOption
70from ..sql.cache_key import HasCacheKey
71from ..sql.operators import ColumnOperators
72from ..sql.schema import Column
73from ..sql.type_api import TypeEngine
74from ..util import warn_deprecated
75from ..util.typing import RODescriptorReference
76from ..util.typing import TupleAny
77from ..util.typing import Unpack
80if typing.TYPE_CHECKING:
81 from ._typing import _EntityType
82 from ._typing import _IdentityKeyType
83 from ._typing import _InstanceDict
84 from ._typing import _InternalEntityType
85 from ._typing import _ORMAdapterProto
86 from .attributes import InstrumentedAttribute
87 from .base import Mapped
88 from .context import _MapperEntity
89 from .context import _ORMCompileState
90 from .context import QueryContext
91 from .decl_api import RegistryType
92 from .decl_base import _ClassScanMapperConfig
93 from .loading import _PopulatorDict
94 from .mapper import Mapper
95 from .path_registry import _AbstractEntityRegistry
96 from .query import Query
97 from .session import Session
98 from .state import InstanceState
99 from .strategy_options import _LoadElement
100 from .util import AliasedInsp
101 from .util import ORMAdapter
102 from ..engine.result import Result
103 from ..sql._typing import _ColumnExpressionArgument
104 from ..sql._typing import _ColumnsClauseArgument
105 from ..sql._typing import _DMLColumnArgument
106 from ..sql._typing import _InfoType
107 from ..sql.operators import OperatorType
108 from ..sql.visitors import _TraverseInternalsType
109 from ..util.typing import _AnnotationScanType
111_StrategyKey = Tuple[Any, ...]
113_T = TypeVar("_T", bound=Any)
114_T_co = TypeVar("_T_co", bound=Any, covariant=True)
116_TLS = TypeVar("_TLS", bound="Type[LoaderStrategy]")
119class ORMStatementRole(roles.StatementRole):
120 __slots__ = ()
121 _role_name = (
122 "Executable SQL or text() construct, including ORM aware objects"
123 )
126class ORMColumnsClauseRole(
127 roles.ColumnsClauseRole, roles.TypedColumnsClauseRole[_T]
128):
129 __slots__ = ()
130 _role_name = "ORM mapped entity, aliased entity, or Column expression"
133class ORMEntityColumnsClauseRole(ORMColumnsClauseRole[_T]):
134 __slots__ = ()
135 _role_name = "ORM mapped or aliased entity"
138class ORMFromClauseRole(roles.FromClauseRole):
139 __slots__ = ()
140 _role_name = "ORM mapped entity, aliased entity, or FROM expression"
143class ORMColumnDescription(TypedDict):
144 name: str
145 # TODO: add python_type and sql_type here; combining them
146 # into "type" is a bad idea
147 type: Union[Type[Any], TypeEngine[Any]]
148 aliased: bool
149 expr: _ColumnsClauseArgument[Any]
150 entity: Optional[_ColumnsClauseArgument[Any]]
153class _IntrospectsAnnotations:
154 __slots__ = ()
156 @classmethod
157 def _mapper_property_name(cls) -> str:
158 return cls.__name__
160 def found_in_pep593_annotated(self) -> Any:
161 """return a copy of this object to use in declarative when the
162 object is found inside of an Annotated object."""
164 raise NotImplementedError(
165 f"Use of the {self._mapper_property_name()!r} "
166 "construct inside of an Annotated object is not yet supported."
167 )
169 def declarative_scan(
170 self,
171 decl_scan: _ClassScanMapperConfig,
172 registry: RegistryType,
173 cls: Type[Any],
174 originating_module: Optional[str],
175 key: str,
176 mapped_container: Optional[Type[Mapped[Any]]],
177 annotation: Optional[_AnnotationScanType],
178 extracted_mapped_annotation: Optional[_AnnotationScanType],
179 is_dataclass_field: bool,
180 ) -> None:
181 """Perform class-specific initializaton at early declarative scanning
182 time.
184 .. versionadded:: 2.0
186 """
188 def _raise_for_required(self, key: str, cls: Type[Any]) -> NoReturn:
189 raise sa_exc.ArgumentError(
190 f"Python typing annotation is required for attribute "
191 f'"{cls.__name__}.{key}" when primary argument(s) for '
192 f'"{self._mapper_property_name()}" '
193 "construct are None or not present"
194 )
197class _DataclassArguments(TypedDict):
198 """define arguments that can be passed to ORM Annotated Dataclass
199 class definitions.
201 """
203 init: Union[_NoArg, bool]
204 repr: Union[_NoArg, bool]
205 eq: Union[_NoArg, bool]
206 order: Union[_NoArg, bool]
207 unsafe_hash: Union[_NoArg, bool]
208 match_args: Union[_NoArg, bool]
209 kw_only: Union[_NoArg, bool]
210 dataclass_callable: Union[_NoArg, Callable[..., Type[Any]]]
213class _AttributeOptions(NamedTuple):
214 """define Python-local attribute behavior options common to all
215 :class:`.MapperProperty` objects.
217 Currently this includes dataclass-generation arguments.
219 .. versionadded:: 2.0
221 """
223 dataclasses_init: Union[_NoArg, bool]
224 dataclasses_repr: Union[_NoArg, bool]
225 dataclasses_default: Union[_NoArg, Any]
226 dataclasses_default_factory: Union[_NoArg, Callable[[], Any]]
227 dataclasses_compare: Union[_NoArg, bool]
228 dataclasses_kw_only: Union[_NoArg, bool]
229 dataclasses_hash: Union[_NoArg, bool, None]
231 def _as_dataclass_field(
232 self, key: str, dataclass_setup_arguments: _DataclassArguments
233 ) -> Any:
234 """Return a ``dataclasses.Field`` object given these arguments."""
236 kw: Dict[str, Any] = {}
237 if self.dataclasses_default_factory is not _NoArg.NO_ARG:
238 kw["default_factory"] = self.dataclasses_default_factory
239 if self.dataclasses_default is not _NoArg.NO_ARG:
240 kw["default"] = self.dataclasses_default
241 if self.dataclasses_init is not _NoArg.NO_ARG:
242 kw["init"] = self.dataclasses_init
243 if self.dataclasses_repr is not _NoArg.NO_ARG:
244 kw["repr"] = self.dataclasses_repr
245 if self.dataclasses_compare is not _NoArg.NO_ARG:
246 kw["compare"] = self.dataclasses_compare
247 if self.dataclasses_kw_only is not _NoArg.NO_ARG:
248 kw["kw_only"] = self.dataclasses_kw_only
249 if self.dataclasses_hash is not _NoArg.NO_ARG:
250 kw["hash"] = self.dataclasses_hash
252 if "default" in kw and callable(kw["default"]):
253 # callable defaults are ambiguous. deprecate them in favour of
254 # insert_default or default_factory. #9936
255 warn_deprecated(
256 f"Callable object passed to the ``default`` parameter for "
257 f"attribute {key!r} in a ORM-mapped Dataclasses context is "
258 "ambiguous, "
259 "and this use will raise an error in a future release. "
260 "If this callable is intended to produce Core level INSERT "
261 "default values for an underlying ``Column``, use "
262 "the ``mapped_column.insert_default`` parameter instead. "
263 "To establish this callable as providing a default value "
264 "for instances of the dataclass itself, use the "
265 "``default_factory`` dataclasses parameter.",
266 "2.0",
267 )
269 if (
270 "init" in kw
271 and not kw["init"]
272 and "default" in kw
273 and not callable(kw["default"]) # ignore callable defaults. #9936
274 and "default_factory" not in kw # illegal but let dc.field raise
275 ):
276 # fix for #9879
277 default = kw.pop("default")
278 kw["default_factory"] = lambda: default
280 return dataclasses.field(**kw)
282 @classmethod
283 def _get_arguments_for_make_dataclass(
284 cls,
285 decl_scan: _ClassScanMapperConfig,
286 key: str,
287 annotation: _AnnotationScanType,
288 mapped_container: Optional[Any],
289 elem: _T,
290 dataclass_setup_arguments: _DataclassArguments,
291 ) -> Union[
292 Tuple[str, _AnnotationScanType],
293 Tuple[str, _AnnotationScanType, dataclasses.Field[Any]],
294 ]:
295 """given attribute key, annotation, and value from a class, return
296 the argument tuple we would pass to dataclasses.make_dataclass()
297 for this attribute.
299 """
300 if isinstance(elem, _DCAttributeOptions):
301 attribute_options = elem._get_dataclass_setup_options(
302 decl_scan, key, dataclass_setup_arguments
303 )
304 dc_field = attribute_options._as_dataclass_field(
305 key, dataclass_setup_arguments
306 )
308 return (key, annotation, dc_field)
309 elif elem is not _NoArg.NO_ARG:
310 # why is typing not erroring on this?
311 return (key, annotation, elem)
312 elif mapped_container is not None:
313 # it's Mapped[], but there's no "element", which means declarative
314 # did not actually do anything for this field. this shouldn't
315 # happen.
316 # previously, this would occur because _scan_attributes would
317 # skip a field that's on an already mapped superclass, but it
318 # would still include it in the annotations, leading
319 # to issue #8718
321 assert False, "Mapped[] received without a mapping declaration"
323 else:
324 # plain dataclass field, not mapped. Is only possible
325 # if __allow_unmapped__ is set up. I can see this mode causing
326 # problems...
327 return (key, annotation)
330_DEFAULT_ATTRIBUTE_OPTIONS = _AttributeOptions(
331 _NoArg.NO_ARG,
332 _NoArg.NO_ARG,
333 _NoArg.NO_ARG,
334 _NoArg.NO_ARG,
335 _NoArg.NO_ARG,
336 _NoArg.NO_ARG,
337 _NoArg.NO_ARG,
338)
340_DEFAULT_READONLY_ATTRIBUTE_OPTIONS = _AttributeOptions(
341 False,
342 _NoArg.NO_ARG,
343 _NoArg.NO_ARG,
344 _NoArg.NO_ARG,
345 _NoArg.NO_ARG,
346 _NoArg.NO_ARG,
347 _NoArg.NO_ARG,
348)
351class _DCAttributeOptions:
352 """mixin for descriptors or configurational objects that include dataclass
353 field options.
355 This includes :class:`.MapperProperty`, :class:`._MapsColumn` within
356 the ORM, but also includes :class:`.AssociationProxy` within ext.
357 Can in theory be used for other descriptors that serve a similar role
358 as association proxy. (*maybe* hybrids, not sure yet.)
360 """
362 __slots__ = ()
364 _attribute_options: _AttributeOptions
365 """behavioral options for ORM-enabled Python attributes
367 .. versionadded:: 2.0
369 """
371 _has_dataclass_arguments: bool
373 def _get_dataclass_setup_options(
374 self,
375 decl_scan: _ClassScanMapperConfig,
376 key: str,
377 dataclass_setup_arguments: _DataclassArguments,
378 ) -> _AttributeOptions:
379 return self._attribute_options
382class _DataclassDefaultsDontSet(_DCAttributeOptions):
383 __slots__ = ()
385 _default_scalar_value: Any
387 _disable_dataclass_default_factory: bool = False
389 def _get_dataclass_setup_options(
390 self,
391 decl_scan: _ClassScanMapperConfig,
392 key: str,
393 dataclass_setup_arguments: _DataclassArguments,
394 ) -> _AttributeOptions:
396 disable_descriptor_defaults = getattr(
397 decl_scan.cls, "_sa_disable_descriptor_defaults", False
398 )
400 dataclasses_default = self._attribute_options.dataclasses_default
401 dataclasses_default_factory = (
402 self._attribute_options.dataclasses_default_factory
403 )
405 if (
406 dataclasses_default is not _NoArg.NO_ARG
407 and not callable(dataclasses_default)
408 and not disable_descriptor_defaults
409 ):
410 self._default_scalar_value = (
411 self._attribute_options.dataclasses_default
412 )
413 return self._attribute_options._replace(
414 dataclasses_default=DONT_SET,
415 )
416 elif (
417 self._disable_dataclass_default_factory
418 and dataclasses_default_factory is not _NoArg.NO_ARG
419 and not disable_descriptor_defaults
420 ):
421 return self._attribute_options._replace(
422 dataclasses_default=DONT_SET,
423 dataclasses_default_factory=_NoArg.NO_ARG,
424 )
425 return self._attribute_options
428class _MapsColumns(_DCAttributeOptions, _MappedAttribute[_T]):
429 """interface for declarative-capable construct that delivers one or more
430 Column objects to the declarative process to be part of a Table.
431 """
433 __slots__ = ()
435 @property
436 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
437 """return a MapperProperty to be assigned to the declarative mapping"""
438 raise NotImplementedError()
440 @property
441 def columns_to_assign(self) -> List[Tuple[Column[_T], int]]:
442 """A list of Column objects that should be declaratively added to the
443 new Table object.
445 """
446 raise NotImplementedError()
449# NOTE: MapperProperty needs to extend _MappedAttribute so that declarative
450# typing works, i.e. "Mapped[A] = relationship()". This introduces an
451# inconvenience which is that all the MapperProperty objects are treated
452# as descriptors by typing tools, which are misled by this as assignment /
453# access to a descriptor attribute wants to move through __get__.
454# Therefore, references to MapperProperty as an instance variable, such
455# as in PropComparator, may have some special typing workarounds such as the
456# use of sqlalchemy.util.typing.DescriptorReference to avoid mis-interpretation
457# by typing tools
458@inspection._self_inspects
459class MapperProperty(
460 HasCacheKey,
461 _DCAttributeOptions,
462 _MappedAttribute[_T],
463 InspectionAttrInfo,
464 util.MemoizedSlots,
465):
466 """Represent a particular class attribute mapped by :class:`_orm.Mapper`.
468 The most common occurrences of :class:`.MapperProperty` are the
469 mapped :class:`_schema.Column`, which is represented in a mapping as
470 an instance of :class:`.ColumnProperty`,
471 and a reference to another class produced by :func:`_orm.relationship`,
472 represented in the mapping as an instance of
473 :class:`.Relationship`.
475 """
477 __slots__ = (
478 "_configure_started",
479 "_configure_finished",
480 "_attribute_options",
481 "_has_dataclass_arguments",
482 "parent",
483 "key",
484 "info",
485 "doc",
486 )
488 _cache_key_traversal: _TraverseInternalsType = [
489 ("parent", visitors.ExtendedInternalTraversal.dp_has_cache_key),
490 ("key", visitors.ExtendedInternalTraversal.dp_string),
491 ]
493 if not TYPE_CHECKING:
494 cascade = None
496 is_property = True
497 """Part of the InspectionAttr interface; states this object is a
498 mapper property.
500 """
502 comparator: PropComparator[_T]
503 """The :class:`_orm.PropComparator` instance that implements SQL
504 expression construction on behalf of this mapped attribute."""
506 key: str
507 """name of class attribute"""
509 parent: Mapper[Any]
510 """the :class:`.Mapper` managing this property."""
512 _is_relationship = False
514 _links_to_entity: bool
515 """True if this MapperProperty refers to a mapped entity.
517 Should only be True for Relationship, False for all others.
519 """
521 doc: Optional[str]
522 """optional documentation string"""
524 info: _InfoType
525 """Info dictionary associated with the object, allowing user-defined
526 data to be associated with this :class:`.InspectionAttr`.
528 The dictionary is generated when first accessed. Alternatively,
529 it can be specified as a constructor argument to the
530 :func:`.column_property`, :func:`_orm.relationship`, or :func:`.composite`
531 functions.
533 .. seealso::
535 :attr:`.QueryableAttribute.info`
537 :attr:`.SchemaItem.info`
539 """
541 def _memoized_attr_info(self) -> _InfoType:
542 """Info dictionary associated with the object, allowing user-defined
543 data to be associated with this :class:`.InspectionAttr`.
545 The dictionary is generated when first accessed. Alternatively,
546 it can be specified as a constructor argument to the
547 :func:`.column_property`, :func:`_orm.relationship`, or
548 :func:`.composite`
549 functions.
551 .. seealso::
553 :attr:`.QueryableAttribute.info`
555 :attr:`.SchemaItem.info`
557 """
558 return {}
560 def setup(
561 self,
562 context: _ORMCompileState,
563 query_entity: _MapperEntity,
564 path: _AbstractEntityRegistry,
565 adapter: Optional[ORMAdapter],
566 **kwargs: Any,
567 ) -> None:
568 """Called by Query for the purposes of constructing a SQL statement.
570 Each MapperProperty associated with the target mapper processes the
571 statement referenced by the query context, adding columns and/or
572 criterion as appropriate.
574 """
576 def create_row_processor(
577 self,
578 context: _ORMCompileState,
579 query_entity: _MapperEntity,
580 path: _AbstractEntityRegistry,
581 mapper: Mapper[Any],
582 result: Result[Unpack[TupleAny]],
583 adapter: Optional[ORMAdapter],
584 populators: _PopulatorDict,
585 ) -> None:
586 """Produce row processing functions and append to the given
587 set of populators lists.
589 """
591 def cascade_iterator(
592 self,
593 type_: str,
594 state: InstanceState[Any],
595 dict_: _InstanceDict,
596 visited_states: Set[InstanceState[Any]],
597 halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
598 ) -> Iterator[
599 Tuple[object, Mapper[Any], InstanceState[Any], _InstanceDict]
600 ]:
601 """Iterate through instances related to the given instance for
602 a particular 'cascade', starting with this MapperProperty.
604 Return an iterator3-tuples (instance, mapper, state).
606 Note that the 'cascade' collection on this MapperProperty is
607 checked first for the given type before cascade_iterator is called.
609 This method typically only applies to Relationship.
611 """
613 return iter(())
615 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
616 """Set the parent mapper that references this MapperProperty.
618 This method is overridden by some subclasses to perform extra
619 setup when the mapper is first known.
621 """
622 self.parent = parent
624 def instrument_class(self, mapper: Mapper[Any]) -> None:
625 """Hook called by the Mapper to the property to initiate
626 instrumentation of the class attribute managed by this
627 MapperProperty.
629 The MapperProperty here will typically call out to the
630 attributes module to set up an InstrumentedAttribute.
632 This step is the first of two steps to set up an InstrumentedAttribute,
633 and is called early in the mapper setup process.
635 The second step is typically the init_class_attribute step,
636 called from StrategizedProperty via the post_instrument_class()
637 hook. This step assigns additional state to the InstrumentedAttribute
638 (specifically the "impl") which has been determined after the
639 MapperProperty has determined what kind of persistence
640 management it needs to do (e.g. scalar, object, collection, etc).
642 """
644 def __init__(
645 self,
646 attribute_options: Optional[_AttributeOptions] = None,
647 _assume_readonly_dc_attributes: bool = False,
648 ) -> None:
649 self._configure_started = False
650 self._configure_finished = False
652 if _assume_readonly_dc_attributes:
653 default_attrs = _DEFAULT_READONLY_ATTRIBUTE_OPTIONS
654 else:
655 default_attrs = _DEFAULT_ATTRIBUTE_OPTIONS
657 if attribute_options and attribute_options != default_attrs:
658 self._has_dataclass_arguments = True
659 self._attribute_options = attribute_options
660 else:
661 self._has_dataclass_arguments = False
662 self._attribute_options = default_attrs
664 def init(self) -> None:
665 """Called after all mappers are created to assemble
666 relationships between mappers and perform other post-mapper-creation
667 initialization steps.
670 """
671 self._configure_started = True
672 self.do_init()
673 self._configure_finished = True
675 @property
676 def class_attribute(self) -> InstrumentedAttribute[_T]:
677 """Return the class-bound descriptor corresponding to this
678 :class:`.MapperProperty`.
680 This is basically a ``getattr()`` call::
682 return getattr(self.parent.class_, self.key)
684 I.e. if this :class:`.MapperProperty` were named ``addresses``,
685 and the class to which it is mapped is ``User``, this sequence
686 is possible::
688 >>> from sqlalchemy import inspect
689 >>> mapper = inspect(User)
690 >>> addresses_property = mapper.attrs.addresses
691 >>> addresses_property.class_attribute is User.addresses
692 True
693 >>> User.addresses.property is addresses_property
694 True
697 """
699 return getattr(self.parent.class_, self.key) # type: ignore
701 def do_init(self) -> None:
702 """Perform subclass-specific initialization post-mapper-creation
703 steps.
705 This is a template method called by the ``MapperProperty``
706 object's init() method.
708 """
710 def post_instrument_class(self, mapper: Mapper[Any]) -> None:
711 """Perform instrumentation adjustments that need to occur
712 after init() has completed.
714 The given Mapper is the Mapper invoking the operation, which
715 may not be the same Mapper as self.parent in an inheritance
716 scenario; however, Mapper will always at least be a sub-mapper of
717 self.parent.
719 This method is typically used by StrategizedProperty, which delegates
720 it to LoaderStrategy.init_class_attribute() to perform final setup
721 on the class-bound InstrumentedAttribute.
723 """
725 def merge(
726 self,
727 session: Session,
728 source_state: InstanceState[Any],
729 source_dict: _InstanceDict,
730 dest_state: InstanceState[Any],
731 dest_dict: _InstanceDict,
732 load: bool,
733 _recursive: Dict[Any, object],
734 _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
735 ) -> None:
736 """Merge the attribute represented by this ``MapperProperty``
737 from source to destination object.
739 """
741 def __repr__(self) -> str:
742 return "<%s at 0x%x; %s>" % (
743 self.__class__.__name__,
744 id(self),
745 getattr(self, "key", "no key"),
746 )
749@inspection._self_inspects
750class PropComparator(SQLORMOperations[_T_co], Generic[_T_co], ColumnOperators):
751 r"""Defines SQL operations for ORM mapped attributes.
753 SQLAlchemy allows for operators to
754 be redefined at both the Core and ORM level. :class:`.PropComparator`
755 is the base class of operator redefinition for ORM-level operations,
756 including those of :class:`.ColumnProperty`,
757 :class:`.Relationship`, and :class:`.Composite`.
759 User-defined subclasses of :class:`.PropComparator` may be created. The
760 built-in Python comparison and math operator methods, such as
761 :meth:`.operators.ColumnOperators.__eq__`,
762 :meth:`.operators.ColumnOperators.__lt__`, and
763 :meth:`.operators.ColumnOperators.__add__`, can be overridden to provide
764 new operator behavior. The custom :class:`.PropComparator` is passed to
765 the :class:`.MapperProperty` instance via the ``comparator_factory``
766 argument. In each case,
767 the appropriate subclass of :class:`.PropComparator` should be used::
769 # definition of custom PropComparator subclasses
771 from sqlalchemy.orm.properties import (
772 ColumnProperty,
773 Composite,
774 Relationship,
775 )
778 class MyColumnComparator(ColumnProperty.Comparator):
779 def __eq__(self, other):
780 return self.__clause_element__() == other
783 class MyRelationshipComparator(Relationship.Comparator):
784 def any(self, expression):
785 "define the 'any' operation"
786 # ...
789 class MyCompositeComparator(Composite.Comparator):
790 def __gt__(self, other):
791 "redefine the 'greater than' operation"
793 return sql.and_(
794 *[
795 a > b
796 for a, b in zip(
797 self.__clause_element__().clauses,
798 other.__composite_values__(),
799 )
800 ]
801 )
804 # application of custom PropComparator subclasses
806 from sqlalchemy.orm import column_property, relationship, composite
807 from sqlalchemy import Column, String
810 class SomeMappedClass(Base):
811 some_column = column_property(
812 Column("some_column", String),
813 comparator_factory=MyColumnComparator,
814 )
816 some_relationship = relationship(
817 SomeOtherClass, comparator_factory=MyRelationshipComparator
818 )
820 some_composite = composite(
821 Column("a", String),
822 Column("b", String),
823 comparator_factory=MyCompositeComparator,
824 )
826 Note that for column-level operator redefinition, it's usually
827 simpler to define the operators at the Core level, using the
828 :attr:`.TypeEngine.comparator_factory` attribute. See
829 :ref:`types_operators` for more detail.
831 .. seealso::
833 :class:`.ColumnProperty.Comparator`
835 :class:`.Relationship.Comparator`
837 :class:`.Composite.Comparator`
839 :class:`.ColumnOperators`
841 :ref:`types_operators`
843 :attr:`.TypeEngine.comparator_factory`
845 """
847 __slots__ = "prop", "_parententity", "_adapt_to_entity"
849 __visit_name__ = "orm_prop_comparator"
851 _parententity: _InternalEntityType[Any]
852 _adapt_to_entity: Optional[AliasedInsp[Any]]
853 prop: RODescriptorReference[MapperProperty[_T_co]]
855 def __init__(
856 self,
857 prop: MapperProperty[_T],
858 parentmapper: _InternalEntityType[Any],
859 adapt_to_entity: Optional[AliasedInsp[Any]] = None,
860 ):
861 self.prop = prop
862 self._parententity = adapt_to_entity or parentmapper
863 self._adapt_to_entity = adapt_to_entity
865 @util.non_memoized_property
866 def property(self) -> MapperProperty[_T_co]:
867 """Return the :class:`.MapperProperty` associated with this
868 :class:`.PropComparator`.
871 Return values here will commonly be instances of
872 :class:`.ColumnProperty` or :class:`.Relationship`.
875 """
876 return self.prop
878 def __clause_element__(self) -> roles.ColumnsClauseRole:
879 raise NotImplementedError("%r" % self)
881 def _bulk_update_tuples(
882 self, value: Any
883 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
884 """Receive a SQL expression that represents a value in the SET
885 clause of an UPDATE statement.
887 Return a tuple that can be passed to a :class:`_expression.Update`
888 construct.
890 """
892 return [(cast("_DMLColumnArgument", self.__clause_element__()), value)]
894 def adapt_to_entity(
895 self, adapt_to_entity: AliasedInsp[Any]
896 ) -> PropComparator[_T_co]:
897 """Return a copy of this PropComparator which will use the given
898 :class:`.AliasedInsp` to produce corresponding expressions.
899 """
900 return self.__class__(self.prop, self._parententity, adapt_to_entity)
902 @util.ro_non_memoized_property
903 def _parentmapper(self) -> Mapper[Any]:
904 """legacy; this is renamed to _parententity to be
905 compatible with QueryableAttribute."""
906 return self._parententity.mapper
908 def _criterion_exists(
909 self,
910 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
911 **kwargs: Any,
912 ) -> ColumnElement[Any]:
913 return self.prop.comparator._criterion_exists(criterion, **kwargs)
915 @util.ro_non_memoized_property
916 def adapter(self) -> Optional[_ORMAdapterProto]:
917 """Produce a callable that adapts column expressions
918 to suit an aliased version of this comparator.
920 """
921 if self._adapt_to_entity is None:
922 return None
923 else:
924 return self._adapt_to_entity._orm_adapt_element
926 @util.ro_non_memoized_property
927 def info(self) -> _InfoType:
928 return self.prop.info
930 @staticmethod
931 def _any_op(a: Any, b: Any, **kwargs: Any) -> Any:
932 return a.any(b, **kwargs)
934 @staticmethod
935 def _has_op(left: Any, other: Any, **kwargs: Any) -> Any:
936 return left.has(other, **kwargs)
938 @staticmethod
939 def _of_type_op(a: Any, class_: Any) -> Any:
940 return a.of_type(class_)
942 any_op = cast(operators.OperatorType, _any_op)
943 has_op = cast(operators.OperatorType, _has_op)
944 of_type_op = cast(operators.OperatorType, _of_type_op)
946 if typing.TYPE_CHECKING:
948 def operate(
949 self, op: OperatorType, *other: Any, **kwargs: Any
950 ) -> ColumnElement[Any]: ...
952 def reverse_operate(
953 self, op: OperatorType, other: Any, **kwargs: Any
954 ) -> ColumnElement[Any]: ...
956 def of_type(self, class_: _EntityType[Any]) -> PropComparator[_T_co]:
957 r"""Redefine this object in terms of a polymorphic subclass,
958 :func:`_orm.with_polymorphic` construct, or :func:`_orm.aliased`
959 construct.
961 Returns a new PropComparator from which further criterion can be
962 evaluated.
964 e.g.::
966 query.join(Company.employees.of_type(Engineer)).filter(
967 Engineer.name == "foo"
968 )
970 :param \class_: a class or mapper indicating that criterion will be
971 against this specific subclass.
973 .. seealso::
975 :ref:`orm_queryguide_joining_relationships_aliased` - in the
976 :ref:`queryguide_toplevel`
978 :ref:`inheritance_of_type`
980 """
982 return self.operate(PropComparator.of_type_op, class_) # type: ignore
984 def and_(
985 self, *criteria: _ColumnExpressionArgument[bool]
986 ) -> PropComparator[bool]:
987 """Add additional criteria to the ON clause that's represented by this
988 relationship attribute.
990 E.g.::
993 stmt = select(User).join(
994 User.addresses.and_(Address.email_address != "foo")
995 )
997 stmt = select(User).options(
998 joinedload(User.addresses.and_(Address.email_address != "foo"))
999 )
1001 .. versionadded:: 1.4
1003 .. seealso::
1005 :ref:`orm_queryguide_join_on_augmented`
1007 :ref:`loader_option_criteria`
1009 :func:`.with_loader_criteria`
1011 """
1012 return self.operate(operators.and_, *criteria) # type: ignore
1014 def any(
1015 self,
1016 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1017 **kwargs: Any,
1018 ) -> ColumnElement[bool]:
1019 r"""Return a SQL expression representing true if this element
1020 references a member which meets the given criterion.
1022 The usual implementation of ``any()`` is
1023 :meth:`.Relationship.Comparator.any`.
1025 :param criterion: an optional ClauseElement formulated against the
1026 member class' table or attributes.
1028 :param \**kwargs: key/value pairs corresponding to member class
1029 attribute names which will be compared via equality to the
1030 corresponding values.
1032 """
1034 return self.operate(PropComparator.any_op, criterion, **kwargs)
1036 def has(
1037 self,
1038 criterion: Optional[_ColumnExpressionArgument[bool]] = None,
1039 **kwargs: Any,
1040 ) -> ColumnElement[bool]:
1041 r"""Return a SQL expression representing true if this element
1042 references a member which meets the given criterion.
1044 The usual implementation of ``has()`` is
1045 :meth:`.Relationship.Comparator.has`.
1047 :param criterion: an optional ClauseElement formulated against the
1048 member class' table or attributes.
1050 :param \**kwargs: key/value pairs corresponding to member class
1051 attribute names which will be compared via equality to the
1052 corresponding values.
1054 """
1056 return self.operate(PropComparator.has_op, criterion, **kwargs)
1059class StrategizedProperty(MapperProperty[_T]):
1060 """A MapperProperty which uses selectable strategies to affect
1061 loading behavior.
1063 There is a single strategy selected by default. Alternate
1064 strategies can be selected at Query time through the usage of
1065 ``StrategizedOption`` objects via the Query.options() method.
1067 The mechanics of StrategizedProperty are used for every Query
1068 invocation for every mapped attribute participating in that Query,
1069 to determine first how the attribute will be rendered in SQL
1070 and secondly how the attribute will retrieve a value from a result
1071 row and apply it to a mapped object. The routines here are very
1072 performance-critical.
1074 """
1076 __slots__ = (
1077 "_strategies",
1078 "strategy",
1079 "_wildcard_token",
1080 "_default_path_loader_key",
1081 "strategy_key",
1082 )
1083 inherit_cache = True
1084 strategy_wildcard_key: ClassVar[str]
1086 strategy_key: _StrategyKey
1088 _strategies: Dict[_StrategyKey, LoaderStrategy]
1090 def _memoized_attr__wildcard_token(self) -> Tuple[str]:
1091 return (
1092 f"{self.strategy_wildcard_key}:{path_registry._WILDCARD_TOKEN}",
1093 )
1095 def _memoized_attr__default_path_loader_key(
1096 self,
1097 ) -> Tuple[str, Tuple[str]]:
1098 return (
1099 "loader",
1100 (f"{self.strategy_wildcard_key}:{path_registry._DEFAULT_TOKEN}",),
1101 )
1103 def _get_context_loader(
1104 self, context: _ORMCompileState, path: _AbstractEntityRegistry
1105 ) -> Optional[_LoadElement]:
1106 load: Optional[_LoadElement] = None
1108 search_path = path[self]
1110 # search among: exact match, "attr.*", "default" strategy
1111 # if any.
1112 for path_key in (
1113 search_path._loader_key,
1114 search_path._wildcard_path_loader_key,
1115 search_path._default_path_loader_key,
1116 ):
1117 if path_key in context.attributes:
1118 load = context.attributes[path_key]
1119 break
1121 # note that if strategy_options.Load is placing non-actionable
1122 # objects in the context like defaultload(), we would
1123 # need to continue the loop here if we got such an
1124 # option as below.
1125 # if load.strategy or load.local_opts:
1126 # break
1128 return load
1130 def _get_strategy(self, key: _StrategyKey) -> LoaderStrategy:
1131 try:
1132 return self._strategies[key]
1133 except KeyError:
1134 pass
1136 # run outside to prevent transfer of exception context
1137 cls = self._strategy_lookup(self, *key)
1138 # this previously was setting self._strategies[cls], that's
1139 # a bad idea; should use strategy key at all times because every
1140 # strategy has multiple keys at this point
1141 self._strategies[key] = strategy = cls(self, key)
1142 return strategy
1144 def setup(
1145 self,
1146 context: _ORMCompileState,
1147 query_entity: _MapperEntity,
1148 path: _AbstractEntityRegistry,
1149 adapter: Optional[ORMAdapter],
1150 **kwargs: Any,
1151 ) -> None:
1152 loader = self._get_context_loader(context, path)
1153 if loader and loader.strategy:
1154 strat = self._get_strategy(loader.strategy)
1155 else:
1156 strat = self.strategy
1157 strat.setup_query(
1158 context, query_entity, path, loader, adapter, **kwargs
1159 )
1161 def create_row_processor(
1162 self,
1163 context: _ORMCompileState,
1164 query_entity: _MapperEntity,
1165 path: _AbstractEntityRegistry,
1166 mapper: Mapper[Any],
1167 result: Result[Unpack[TupleAny]],
1168 adapter: Optional[ORMAdapter],
1169 populators: _PopulatorDict,
1170 ) -> None:
1171 loader = self._get_context_loader(context, path)
1172 if loader and loader.strategy:
1173 strat = self._get_strategy(loader.strategy)
1174 else:
1175 strat = self.strategy
1176 strat.create_row_processor(
1177 context,
1178 query_entity,
1179 path,
1180 loader,
1181 mapper,
1182 result,
1183 adapter,
1184 populators,
1185 )
1187 def do_init(self) -> None:
1188 self._strategies = {}
1189 self.strategy = self._get_strategy(self.strategy_key)
1191 def post_instrument_class(self, mapper: Mapper[Any]) -> None:
1192 if not mapper.class_manager._attr_has_impl(self.key):
1193 self.strategy.init_class_attribute(mapper)
1195 _all_strategies: collections.defaultdict[
1196 Type[MapperProperty[Any]], Dict[_StrategyKey, Type[LoaderStrategy]]
1197 ] = collections.defaultdict(dict)
1199 @classmethod
1200 def strategy_for(cls, **kw: Any) -> Callable[[_TLS], _TLS]:
1201 def decorate(dec_cls: _TLS) -> _TLS:
1202 # ensure each subclass of the strategy has its
1203 # own _strategy_keys collection
1204 if "_strategy_keys" not in dec_cls.__dict__:
1205 dec_cls._strategy_keys = []
1206 key = tuple(sorted(kw.items()))
1207 cls._all_strategies[cls][key] = dec_cls
1208 dec_cls._strategy_keys.append(key)
1209 return dec_cls
1211 return decorate
1213 @classmethod
1214 def _strategy_lookup(
1215 cls, requesting_property: MapperProperty[Any], *key: Any
1216 ) -> Type[LoaderStrategy]:
1217 requesting_property.parent._with_polymorphic_mappers
1219 for prop_cls in cls.__mro__:
1220 if prop_cls in cls._all_strategies:
1221 if TYPE_CHECKING:
1222 assert issubclass(prop_cls, MapperProperty)
1223 strategies = cls._all_strategies[prop_cls]
1224 try:
1225 return strategies[key]
1226 except KeyError:
1227 pass
1229 for property_type, strats in cls._all_strategies.items():
1230 if key in strats:
1231 intended_property_type = property_type
1232 actual_strategy = strats[key]
1233 break
1234 else:
1235 intended_property_type = None
1236 actual_strategy = None
1238 raise orm_exc.LoaderStrategyException(
1239 cls,
1240 requesting_property,
1241 intended_property_type,
1242 actual_strategy,
1243 key,
1244 )
1247class ORMOption(ExecutableOption):
1248 """Base class for option objects that are passed to ORM queries.
1250 These options may be consumed by :meth:`.Query.options`,
1251 :meth:`.Select.options`, or in a more general sense by any
1252 :meth:`.Executable.options` method. They are interpreted at
1253 statement compile time or execution time in modern use. The
1254 deprecated :class:`.MapperOption` is consumed at ORM query construction
1255 time.
1257 .. versionadded:: 1.4
1259 """
1261 __slots__ = ()
1263 _is_legacy_option = False
1265 propagate_to_loaders = False
1266 """if True, indicate this option should be carried along
1267 to "secondary" SELECT statements that occur for relationship
1268 lazy loaders as well as attribute load / refresh operations.
1270 """
1272 _is_core = False
1274 _is_user_defined = False
1276 _is_compile_state = False
1278 _is_criteria_option = False
1280 _is_strategy_option = False
1282 def _adapt_cached_option_to_uncached_option(
1283 self, context: QueryContext, uncached_opt: ORMOption
1284 ) -> ORMOption:
1285 """adapt this option to the "uncached" version of itself in a
1286 loader strategy context.
1288 given "self" which is an option from a cached query, as well as the
1289 corresponding option from the uncached version of the same query,
1290 return the option we should use in a new query, in the context of a
1291 loader strategy being asked to load related rows on behalf of that
1292 cached query, which is assumed to be building a new query based on
1293 entities passed to us from the cached query.
1295 Currently this routine chooses between "self" and "uncached" without
1296 manufacturing anything new. If the option is itself a loader strategy
1297 option which has a path, that path needs to match to the entities being
1298 passed to us by the cached query, so the :class:`_orm.Load` subclass
1299 overrides this to return "self". For all other options, we return the
1300 uncached form which may have changing state, such as a
1301 with_loader_criteria() option which will very often have new state.
1303 This routine could in the future involve
1304 generating a new option based on both inputs if use cases arise,
1305 such as if with_loader_criteria() needed to match up to
1306 ``AliasedClass`` instances given in the parent query.
1308 However, longer term it might be better to restructure things such that
1309 ``AliasedClass`` entities are always matched up on their cache key,
1310 instead of identity, in things like paths and such, so that this whole
1311 issue of "the uncached option does not match the entities" goes away.
1312 However this would make ``PathRegistry`` more complicated and difficult
1313 to debug as well as potentially less performant in that it would be
1314 hashing enormous cache keys rather than a simple AliasedInsp. UNLESS,
1315 we could get cache keys overall to be reliably hashed into something
1316 like an md5 key.
1318 .. versionadded:: 1.4.41
1320 """
1321 if uncached_opt is not None:
1322 return uncached_opt
1323 else:
1324 return self
1327class CompileStateOption(HasCacheKey, ORMOption):
1328 """base for :class:`.ORMOption` classes that affect the compilation of
1329 a SQL query and therefore need to be part of the cache key.
1331 .. note:: :class:`.CompileStateOption` is generally non-public and
1332 should not be used as a base class for user-defined options; instead,
1333 use :class:`.UserDefinedOption`, which is easier to use as it does not
1334 interact with ORM compilation internals or caching.
1336 :class:`.CompileStateOption` defines an internal attribute
1337 ``_is_compile_state=True`` which has the effect of the ORM compilation
1338 routines for SELECT and other statements will call upon these options when
1339 a SQL string is being compiled. As such, these classes implement
1340 :class:`.HasCacheKey` and need to provide robust ``_cache_key_traversal``
1341 structures.
1343 The :class:`.CompileStateOption` class is used to implement the ORM
1344 :class:`.LoaderOption` and :class:`.CriteriaOption` classes.
1346 .. versionadded:: 1.4.28
1349 """
1351 __slots__ = ()
1353 _is_compile_state = True
1355 def process_compile_state(self, compile_state: _ORMCompileState) -> None:
1356 """Apply a modification to a given :class:`.ORMCompileState`.
1358 This method is part of the implementation of a particular
1359 :class:`.CompileStateOption` and is only invoked internally
1360 when an ORM query is compiled.
1362 """
1364 def process_compile_state_replaced_entities(
1365 self,
1366 compile_state: _ORMCompileState,
1367 mapper_entities: Sequence[_MapperEntity],
1368 ) -> None:
1369 """Apply a modification to a given :class:`.ORMCompileState`,
1370 given entities that were replaced by with_only_columns() or
1371 with_entities().
1373 This method is part of the implementation of a particular
1374 :class:`.CompileStateOption` and is only invoked internally
1375 when an ORM query is compiled.
1377 .. versionadded:: 1.4.19
1379 """
1382class LoaderOption(CompileStateOption):
1383 """Describe a loader modification to an ORM statement at compilation time.
1385 .. versionadded:: 1.4
1387 """
1389 __slots__ = ()
1391 def process_compile_state_replaced_entities(
1392 self,
1393 compile_state: _ORMCompileState,
1394 mapper_entities: Sequence[_MapperEntity],
1395 ) -> None:
1396 self.process_compile_state(compile_state)
1399class CriteriaOption(CompileStateOption):
1400 """Describe a WHERE criteria modification to an ORM statement at
1401 compilation time.
1403 .. versionadded:: 1.4
1405 """
1407 __slots__ = ()
1409 _is_criteria_option = True
1411 def get_global_criteria(self, attributes: Dict[str, Any]) -> None:
1412 """update additional entity criteria options in the given
1413 attributes dictionary.
1415 """
1418class UserDefinedOption(ORMOption):
1419 """Base class for a user-defined option that can be consumed from the
1420 :meth:`.SessionEvents.do_orm_execute` event hook.
1422 """
1424 __slots__ = ("payload",)
1426 _is_legacy_option = False
1428 _is_user_defined = True
1430 propagate_to_loaders = False
1431 """if True, indicate this option should be carried along
1432 to "secondary" Query objects produced during lazy loads
1433 or refresh operations.
1435 """
1437 def __init__(self, payload: Optional[Any] = None):
1438 self.payload = payload
1441@util.deprecated_cls(
1442 "1.4",
1443 "The :class:`.MapperOption class is deprecated and will be removed "
1444 "in a future release. For "
1445 "modifications to queries on a per-execution basis, use the "
1446 ":class:`.UserDefinedOption` class to establish state within a "
1447 ":class:`.Query` or other Core statement, then use the "
1448 ":meth:`.SessionEvents.before_orm_execute` hook to consume them.",
1449 constructor=None,
1450)
1451class MapperOption(ORMOption):
1452 """Describe a modification to a Query"""
1454 __slots__ = ()
1456 _is_legacy_option = True
1458 propagate_to_loaders = False
1459 """if True, indicate this option should be carried along
1460 to "secondary" Query objects produced during lazy loads
1461 or refresh operations.
1463 """
1465 def process_query(self, query: Query[Any]) -> None:
1466 """Apply a modification to the given :class:`_query.Query`."""
1468 def process_query_conditionally(self, query: Query[Any]) -> None:
1469 """same as process_query(), except that this option may not
1470 apply to the given query.
1472 This is typically applied during a lazy load or scalar refresh
1473 operation to propagate options stated in the original Query to the
1474 new Query being used for the load. It occurs for those options that
1475 specify propagate_to_loaders=True.
1477 """
1479 self.process_query(query)
1482class LoaderStrategy:
1483 """Describe the loading behavior of a StrategizedProperty object.
1485 The ``LoaderStrategy`` interacts with the querying process in three
1486 ways:
1488 * it controls the configuration of the ``InstrumentedAttribute``
1489 placed on a class to handle the behavior of the attribute. this
1490 may involve setting up class-level callable functions to fire
1491 off a select operation when the attribute is first accessed
1492 (i.e. a lazy load)
1494 * it processes the ``QueryContext`` at statement construction time,
1495 where it can modify the SQL statement that is being produced.
1496 For example, simple column attributes will add their represented
1497 column to the list of selected columns, a joined eager loader
1498 may establish join clauses to add to the statement.
1500 * It produces "row processor" functions at result fetching time.
1501 These "row processor" functions populate a particular attribute
1502 on a particular mapped instance.
1504 """
1506 __slots__ = (
1507 "parent_property",
1508 "is_class_level",
1509 "parent",
1510 "key",
1511 "strategy_key",
1512 "strategy_opts",
1513 )
1515 _strategy_keys: ClassVar[List[_StrategyKey]]
1517 def __init__(
1518 self, parent: MapperProperty[Any], strategy_key: _StrategyKey
1519 ):
1520 self.parent_property = parent
1521 self.is_class_level = False
1522 self.parent = self.parent_property.parent
1523 self.key = self.parent_property.key
1524 self.strategy_key = strategy_key
1525 self.strategy_opts = dict(strategy_key)
1527 def init_class_attribute(self, mapper: Mapper[Any]) -> None:
1528 pass
1530 def setup_query(
1531 self,
1532 compile_state: _ORMCompileState,
1533 query_entity: _MapperEntity,
1534 path: _AbstractEntityRegistry,
1535 loadopt: Optional[_LoadElement],
1536 adapter: Optional[ORMAdapter],
1537 **kwargs: Any,
1538 ) -> None:
1539 """Establish column and other state for a given QueryContext.
1541 This method fulfills the contract specified by MapperProperty.setup().
1543 StrategizedProperty delegates its setup() method
1544 directly to this method.
1546 """
1548 def create_row_processor(
1549 self,
1550 context: _ORMCompileState,
1551 query_entity: _MapperEntity,
1552 path: _AbstractEntityRegistry,
1553 loadopt: Optional[_LoadElement],
1554 mapper: Mapper[Any],
1555 result: Result[Unpack[TupleAny]],
1556 adapter: Optional[ORMAdapter],
1557 populators: _PopulatorDict,
1558 ) -> None:
1559 """Establish row processing functions for a given QueryContext.
1561 This method fulfills the contract specified by
1562 MapperProperty.create_row_processor().
1564 StrategizedProperty delegates its create_row_processor() method
1565 directly to this method.
1567 """
1569 def __str__(self) -> str:
1570 return str(self.parent_property)