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/descriptor_props.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"""Descriptor properties are more "auxiliary" properties
9that exist as configurational elements, but don't participate
10as actively in the load/persist ORM loop.
12"""
13from __future__ import annotations
15from dataclasses import is_dataclass
16import inspect
17import itertools
18import operator
19import typing
20from typing import Any
21from typing import Callable
22from typing import Dict
23from typing import List
24from typing import NoReturn
25from typing import Optional
26from typing import Sequence
27from typing import Tuple
28from typing import Type
29from typing import TYPE_CHECKING
30from typing import TypeVar
31from typing import Union
32import weakref
34from . import attributes
35from . import util as orm_util
36from .base import _DeclarativeMapped
37from .base import DONT_SET
38from .base import LoaderCallableStatus
39from .base import Mapped
40from .base import PassiveFlag
41from .base import SQLORMOperations
42from .interfaces import _AttributeOptions
43from .interfaces import _IntrospectsAnnotations
44from .interfaces import _MapsColumns
45from .interfaces import MapperProperty
46from .interfaces import PropComparator
47from .util import _none_set
48from .util import de_stringify_annotation
49from .. import event
50from .. import exc as sa_exc
51from .. import schema
52from .. import sql
53from .. import util
54from ..sql import expression
55from ..sql import operators
56from ..sql.base import _NoArg
57from ..sql.elements import BindParameter
58from ..util.typing import get_args
59from ..util.typing import is_fwd_ref
60from ..util.typing import is_pep593
61from ..util.typing import TupleAny
62from ..util.typing import Unpack
65if typing.TYPE_CHECKING:
66 from ._typing import _InstanceDict
67 from ._typing import _RegistryType
68 from .attributes import History
69 from .attributes import InstrumentedAttribute
70 from .attributes import QueryableAttribute
71 from .context import _ORMCompileState
72 from .decl_base import _ClassScanMapperConfig
73 from .interfaces import _DataclassArguments
74 from .mapper import Mapper
75 from .properties import ColumnProperty
76 from .properties import MappedColumn
77 from .state import InstanceState
78 from ..engine.base import Connection
79 from ..engine.row import Row
80 from ..sql._typing import _DMLColumnArgument
81 from ..sql._typing import _InfoType
82 from ..sql.elements import ClauseList
83 from ..sql.elements import ColumnElement
84 from ..sql.operators import OperatorType
85 from ..sql.schema import Column
86 from ..sql.selectable import Select
87 from ..util.typing import _AnnotationScanType
88 from ..util.typing import CallableReference
89 from ..util.typing import DescriptorReference
90 from ..util.typing import RODescriptorReference
92_T = TypeVar("_T", bound=Any)
93_PT = TypeVar("_PT", bound=Any)
96class DescriptorProperty(MapperProperty[_T]):
97 """:class:`.MapperProperty` which proxies access to a
98 user-defined descriptor."""
100 doc: Optional[str] = None
102 uses_objects = False
103 _links_to_entity = False
105 descriptor: DescriptorReference[Any]
107 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
108 raise NotImplementedError(
109 "This MapperProperty does not implement column loader strategies"
110 )
112 def get_history(
113 self,
114 state: InstanceState[Any],
115 dict_: _InstanceDict,
116 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
117 ) -> History:
118 raise NotImplementedError()
120 def instrument_class(self, mapper: Mapper[Any]) -> None:
121 prop = self
123 class _ProxyImpl(attributes._AttributeImpl):
124 accepts_scalar_loader = False
125 load_on_unexpire = True
126 collection = False
128 @property
129 def uses_objects(self) -> bool: # type: ignore
130 return prop.uses_objects
132 def __init__(self, key: str):
133 self.key = key
135 def get_history(
136 self,
137 state: InstanceState[Any],
138 dict_: _InstanceDict,
139 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
140 ) -> History:
141 return prop.get_history(state, dict_, passive)
143 if self.descriptor is None:
144 desc = getattr(mapper.class_, self.key, None)
145 if mapper._is_userland_descriptor(self.key, desc):
146 self.descriptor = desc
148 if self.descriptor is None:
150 def fset(obj: Any, value: Any) -> None:
151 setattr(obj, self.name, value)
153 def fdel(obj: Any) -> None:
154 delattr(obj, self.name)
156 def fget(obj: Any) -> Any:
157 return getattr(obj, self.name)
159 self.descriptor = property(fget=fget, fset=fset, fdel=fdel)
161 proxy_attr = attributes._create_proxied_attribute(self.descriptor)(
162 self.parent.class_,
163 self.key,
164 self.descriptor,
165 lambda: self._comparator_factory(mapper),
166 doc=self.doc,
167 original_property=self,
168 )
170 proxy_attr.impl = _ProxyImpl(self.key)
171 mapper.class_manager.instrument_attribute(self.key, proxy_attr)
174_CompositeAttrType = Union[
175 str,
176 "Column[_T]",
177 "MappedColumn[_T]",
178 "InstrumentedAttribute[_T]",
179 "Mapped[_T]",
180]
183_CC = TypeVar("_CC", bound=Any)
186_composite_getters: weakref.WeakKeyDictionary[
187 Type[Any], Callable[[Any], Tuple[Any, ...]]
188] = weakref.WeakKeyDictionary()
191class CompositeProperty(
192 _MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC]
193):
194 """Defines a "composite" mapped attribute, representing a collection
195 of columns as one attribute.
197 :class:`.CompositeProperty` is constructed using the :func:`.composite`
198 function.
200 .. seealso::
202 :ref:`mapper_composite`
204 """
206 composite_class: Union[Type[_CC], Callable[..., _CC]]
207 attrs: Tuple[_CompositeAttrType[Any], ...]
209 _generated_composite_accessor: CallableReference[
210 Optional[Callable[[_CC], Tuple[Any, ...]]]
211 ]
213 comparator_factory: Type[Comparator[_CC]]
215 def __init__(
216 self,
217 _class_or_attr: Union[
218 None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any]
219 ] = None,
220 *attrs: _CompositeAttrType[Any],
221 attribute_options: Optional[_AttributeOptions] = None,
222 active_history: bool = False,
223 deferred: bool = False,
224 group: Optional[str] = None,
225 comparator_factory: Optional[Type[Comparator[_CC]]] = None,
226 info: Optional[_InfoType] = None,
227 **kwargs: Any,
228 ):
229 super().__init__(attribute_options=attribute_options)
231 if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
232 self.attrs = (_class_or_attr,) + attrs
233 # will initialize within declarative_scan
234 self.composite_class = None # type: ignore
235 else:
236 self.composite_class = _class_or_attr # type: ignore
237 self.attrs = attrs
239 self.active_history = active_history
240 self.deferred = deferred
241 self.group = group
242 self.comparator_factory = (
243 comparator_factory
244 if comparator_factory is not None
245 else self.__class__.Comparator
246 )
247 self._generated_composite_accessor = None
248 if info is not None:
249 self.info.update(info)
251 util.set_creation_order(self)
252 self._create_descriptor()
253 self._init_accessor()
255 def instrument_class(self, mapper: Mapper[Any]) -> None:
256 super().instrument_class(mapper)
257 self._setup_event_handlers()
259 def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]:
260 if self._generated_composite_accessor:
261 return self._generated_composite_accessor(value)
262 else:
263 try:
264 accessor = value.__composite_values__
265 except AttributeError as ae:
266 raise sa_exc.InvalidRequestError(
267 f"Composite class {self.composite_class.__name__} is not "
268 f"a dataclass and does not define a __composite_values__()"
269 " method; can't get state"
270 ) from ae
271 else:
272 return accessor() # type: ignore
274 def do_init(self) -> None:
275 """Initialization which occurs after the :class:`.Composite`
276 has been associated with its parent mapper.
278 """
279 self._setup_arguments_on_columns()
281 _COMPOSITE_FGET = object()
283 def _create_descriptor(self) -> None:
284 """Create the Python descriptor that will serve as
285 the access point on instances of the mapped class.
287 """
289 def fget(instance: Any) -> Any:
290 dict_ = attributes.instance_dict(instance)
291 state = attributes.instance_state(instance)
293 if self.key not in dict_:
294 # key not present. Iterate through related
295 # attributes, retrieve their values. This
296 # ensures they all load.
297 values = [
298 getattr(instance, key) for key in self._attribute_keys
299 ]
301 # current expected behavior here is that the composite is
302 # created on access if the object is persistent or if
303 # col attributes have non-None. This would be better
304 # if the composite were created unconditionally,
305 # but that would be a behavioral change.
306 if self.key not in dict_ and (
307 state.key is not None or not _none_set.issuperset(values)
308 ):
309 dict_[self.key] = self.composite_class(*values)
310 state.manager.dispatch.refresh(
311 state, self._COMPOSITE_FGET, [self.key]
312 )
314 return dict_.get(self.key, None)
316 def fset(instance: Any, value: Any) -> None:
317 if value is LoaderCallableStatus.DONT_SET:
318 return
320 dict_ = attributes.instance_dict(instance)
321 state = attributes.instance_state(instance)
322 attr = state.manager[self.key]
324 if attr.dispatch._active_history:
325 previous = fget(instance)
326 else:
327 previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
329 for fn in attr.dispatch.set:
330 value = fn(state, value, previous, attr.impl)
331 dict_[self.key] = value
332 if value is None:
333 for key in self._attribute_keys:
334 setattr(instance, key, None)
335 else:
336 for key, value in zip(
337 self._attribute_keys,
338 self._composite_values_from_instance(value),
339 ):
340 setattr(instance, key, value)
342 def fdel(instance: Any) -> None:
343 state = attributes.instance_state(instance)
344 dict_ = attributes.instance_dict(instance)
345 attr = state.manager[self.key]
347 if attr.dispatch._active_history:
348 previous = fget(instance)
349 dict_.pop(self.key, None)
350 else:
351 previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
353 attr = state.manager[self.key]
354 attr.dispatch.remove(state, previous, attr.impl)
355 for key in self._attribute_keys:
356 setattr(instance, key, None)
358 self.descriptor = property(fget, fset, fdel)
360 @util.preload_module("sqlalchemy.orm.properties")
361 def declarative_scan(
362 self,
363 decl_scan: _ClassScanMapperConfig,
364 registry: _RegistryType,
365 cls: Type[Any],
366 originating_module: Optional[str],
367 key: str,
368 mapped_container: Optional[Type[Mapped[Any]]],
369 annotation: Optional[_AnnotationScanType],
370 extracted_mapped_annotation: Optional[_AnnotationScanType],
371 is_dataclass_field: bool,
372 ) -> None:
373 MappedColumn = util.preloaded.orm_properties.MappedColumn
374 if (
375 self.composite_class is None
376 and extracted_mapped_annotation is None
377 ):
378 self._raise_for_required(key, cls)
379 argument = extracted_mapped_annotation
381 if is_pep593(argument):
382 argument = get_args(argument)[0]
384 if argument and self.composite_class is None:
385 if isinstance(argument, str) or is_fwd_ref(
386 argument, check_generic=True
387 ):
388 if originating_module is None:
389 str_arg = (
390 argument.__forward_arg__
391 if hasattr(argument, "__forward_arg__")
392 else str(argument)
393 )
394 raise sa_exc.ArgumentError(
395 f"Can't use forward ref {argument} for composite "
396 f"class argument; set up the type as Mapped[{str_arg}]"
397 )
398 argument = de_stringify_annotation(
399 cls, argument, originating_module, include_generic=True
400 )
402 self.composite_class = argument
404 if is_dataclass(self.composite_class):
405 self._setup_for_dataclass(registry, cls, originating_module, key)
406 else:
407 for attr in self.attrs:
408 if (
409 isinstance(attr, (MappedColumn, schema.Column))
410 and attr.name is None
411 ):
412 raise sa_exc.ArgumentError(
413 "Composite class column arguments must be named "
414 "unless a dataclass is used"
415 )
416 self._init_accessor()
418 def _init_accessor(self) -> None:
419 if is_dataclass(self.composite_class) and not hasattr(
420 self.composite_class, "__composite_values__"
421 ):
422 insp = inspect.signature(self.composite_class)
423 getter = operator.attrgetter(
424 *[p.name for p in insp.parameters.values()]
425 )
426 if len(insp.parameters) == 1:
427 self._generated_composite_accessor = lambda obj: (getter(obj),)
428 else:
429 self._generated_composite_accessor = getter
431 if (
432 self.composite_class is not None
433 and isinstance(self.composite_class, type)
434 and self.composite_class not in _composite_getters
435 ):
436 if self._generated_composite_accessor is not None:
437 _composite_getters[self.composite_class] = (
438 self._generated_composite_accessor
439 )
440 elif hasattr(self.composite_class, "__composite_values__"):
441 _composite_getters[self.composite_class] = (
442 lambda obj: obj.__composite_values__()
443 )
445 @util.preload_module("sqlalchemy.orm.properties")
446 @util.preload_module("sqlalchemy.orm.decl_base")
447 def _setup_for_dataclass(
448 self,
449 registry: _RegistryType,
450 cls: Type[Any],
451 originating_module: Optional[str],
452 key: str,
453 ) -> None:
454 MappedColumn = util.preloaded.orm_properties.MappedColumn
456 decl_base = util.preloaded.orm_decl_base
458 insp = inspect.signature(self.composite_class)
459 for param, attr in itertools.zip_longest(
460 insp.parameters.values(), self.attrs
461 ):
462 if param is None:
463 raise sa_exc.ArgumentError(
464 f"number of composite attributes "
465 f"{len(self.attrs)} exceeds "
466 f"that of the number of attributes in class "
467 f"{self.composite_class.__name__} {len(insp.parameters)}"
468 )
469 if attr is None:
470 # fill in missing attr spots with empty MappedColumn
471 attr = MappedColumn()
472 self.attrs += (attr,)
474 if isinstance(attr, MappedColumn):
475 attr.declarative_scan_for_composite(
476 registry,
477 cls,
478 originating_module,
479 key,
480 param.name,
481 param.annotation,
482 )
483 elif isinstance(attr, schema.Column):
484 decl_base._undefer_column_name(param.name, attr)
486 @util.memoized_property
487 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
488 return [getattr(self.parent.class_, prop.key) for prop in self.props]
490 @util.memoized_property
491 @util.preload_module("orm.properties")
492 def props(self) -> Sequence[MapperProperty[Any]]:
493 props = []
494 MappedColumn = util.preloaded.orm_properties.MappedColumn
496 for attr in self.attrs:
497 if isinstance(attr, str):
498 prop = self.parent.get_property(attr, _configure_mappers=False)
499 elif isinstance(attr, schema.Column):
500 prop = self.parent._columntoproperty[attr]
501 elif isinstance(attr, MappedColumn):
502 prop = self.parent._columntoproperty[attr.column]
503 elif isinstance(attr, attributes.InstrumentedAttribute):
504 prop = attr.property
505 else:
506 prop = None
508 if not isinstance(prop, MapperProperty):
509 raise sa_exc.ArgumentError(
510 "Composite expects Column objects or mapped "
511 f"attributes/attribute names as arguments, got: {attr!r}"
512 )
514 props.append(prop)
515 return props
517 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
518 return self._comparable_elements
520 @util.non_memoized_property
521 @util.preload_module("orm.properties")
522 def columns(self) -> Sequence[Column[Any]]:
523 MappedColumn = util.preloaded.orm_properties.MappedColumn
524 return [
525 a.column if isinstance(a, MappedColumn) else a
526 for a in self.attrs
527 if isinstance(a, (schema.Column, MappedColumn))
528 ]
530 @property
531 def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]:
532 return self
534 @property
535 def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
536 return [(c, 0) for c in self.columns if c.table is None]
538 @util.preload_module("orm.properties")
539 def _setup_arguments_on_columns(self) -> None:
540 """Propagate configuration arguments made on this composite
541 to the target columns, for those that apply.
543 """
544 ColumnProperty = util.preloaded.orm_properties.ColumnProperty
546 for prop in self.props:
547 if not isinstance(prop, ColumnProperty):
548 continue
549 else:
550 cprop = prop
552 cprop.active_history = self.active_history
553 if self.deferred:
554 cprop.deferred = self.deferred
555 cprop.strategy_key = (("deferred", True), ("instrument", True))
556 cprop.group = self.group
558 def _setup_event_handlers(self) -> None:
559 """Establish events that populate/expire the composite attribute."""
561 def load_handler(
562 state: InstanceState[Any], context: _ORMCompileState
563 ) -> None:
564 _load_refresh_handler(state, context, None, is_refresh=False)
566 def refresh_handler(
567 state: InstanceState[Any],
568 context: _ORMCompileState,
569 to_load: Optional[Sequence[str]],
570 ) -> None:
571 # note this corresponds to sqlalchemy.ext.mutable load_attrs()
573 if not to_load or (
574 {self.key}.union(self._attribute_keys)
575 ).intersection(to_load):
576 _load_refresh_handler(state, context, to_load, is_refresh=True)
578 def _load_refresh_handler(
579 state: InstanceState[Any],
580 context: _ORMCompileState,
581 to_load: Optional[Sequence[str]],
582 is_refresh: bool,
583 ) -> None:
584 dict_ = state.dict
586 # if context indicates we are coming from the
587 # fget() handler, this already set the value; skip the
588 # handler here. (other handlers like mutablecomposite will still
589 # want to catch it)
590 # there's an insufficiency here in that the fget() handler
591 # really should not be using the refresh event and there should
592 # be some other event that mutablecomposite can subscribe
593 # towards for this.
595 if (
596 not is_refresh or context is self._COMPOSITE_FGET
597 ) and self.key in dict_:
598 return
600 # if column elements aren't loaded, skip.
601 # __get__() will initiate a load for those
602 # columns
603 for k in self._attribute_keys:
604 if k not in dict_:
605 return
607 dict_[self.key] = self.composite_class(
608 *[state.dict[key] for key in self._attribute_keys]
609 )
611 def expire_handler(
612 state: InstanceState[Any], keys: Optional[Sequence[str]]
613 ) -> None:
614 if keys is None or set(self._attribute_keys).intersection(keys):
615 state.dict.pop(self.key, None)
617 def insert_update_handler(
618 mapper: Mapper[Any],
619 connection: Connection,
620 state: InstanceState[Any],
621 ) -> None:
622 """After an insert or update, some columns may be expired due
623 to server side defaults, or re-populated due to client side
624 defaults. Pop out the composite value here so that it
625 recreates.
627 """
629 state.dict.pop(self.key, None)
631 event.listen(
632 self.parent, "after_insert", insert_update_handler, raw=True
633 )
634 event.listen(
635 self.parent, "after_update", insert_update_handler, raw=True
636 )
637 event.listen(
638 self.parent, "load", load_handler, raw=True, propagate=True
639 )
640 event.listen(
641 self.parent, "refresh", refresh_handler, raw=True, propagate=True
642 )
643 event.listen(
644 self.parent, "expire", expire_handler, raw=True, propagate=True
645 )
647 proxy_attr = self.parent.class_manager[self.key]
648 proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore
649 proxy_attr.impl.dispatch._active_history = self.active_history
651 # TODO: need a deserialize hook here
653 @util.memoized_property
654 def _attribute_keys(self) -> Sequence[str]:
655 return [prop.key for prop in self.props]
657 def _populate_composite_bulk_save_mappings_fn(
658 self,
659 ) -> Callable[[Dict[str, Any]], None]:
660 if self._generated_composite_accessor:
661 get_values = self._generated_composite_accessor
662 else:
664 def get_values(val: Any) -> Tuple[Any]:
665 return val.__composite_values__() # type: ignore
667 attrs = [prop.key for prop in self.props]
669 def populate(dest_dict: Dict[str, Any]) -> None:
670 dest_dict.update(
671 {
672 key: val
673 for key, val in zip(
674 attrs, get_values(dest_dict.pop(self.key))
675 )
676 }
677 )
679 return populate
681 def get_history(
682 self,
683 state: InstanceState[Any],
684 dict_: _InstanceDict,
685 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
686 ) -> History:
687 """Provided for userland code that uses attributes.get_history()."""
689 added: List[Any] = []
690 deleted: List[Any] = []
692 has_history = False
693 for prop in self.props:
694 key = prop.key
695 hist = state.manager[key].impl.get_history(state, dict_)
696 if hist.has_changes():
697 has_history = True
699 non_deleted = hist.non_deleted()
700 if non_deleted:
701 added.extend(non_deleted)
702 else:
703 added.append(None)
704 if hist.deleted:
705 deleted.extend(hist.deleted)
706 else:
707 deleted.append(None)
709 if has_history:
710 return attributes.History(
711 [self.composite_class(*added)],
712 (),
713 [self.composite_class(*deleted)],
714 )
715 else:
716 return attributes.History((), [self.composite_class(*added)], ())
718 def _comparator_factory(
719 self, mapper: Mapper[Any]
720 ) -> Composite.Comparator[_CC]:
721 return self.comparator_factory(self, mapper)
723 class CompositeBundle(orm_util.Bundle[_T]):
724 def __init__(
725 self,
726 property_: Composite[_T],
727 expr: ClauseList,
728 ):
729 self.property = property_
730 super().__init__(property_.key, *expr)
732 def create_row_processor(
733 self,
734 query: Select[Unpack[TupleAny]],
735 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
736 labels: Sequence[str],
737 ) -> Callable[[Row[Unpack[TupleAny]]], Any]:
738 def proc(row: Row[Unpack[TupleAny]]) -> Any:
739 return self.property.composite_class(
740 *[proc(row) for proc in procs]
741 )
743 return proc
745 class Comparator(PropComparator[_PT]):
746 """Produce boolean, comparison, and other operators for
747 :class:`.Composite` attributes.
749 See the example in :ref:`composite_operations` for an overview
750 of usage , as well as the documentation for :class:`.PropComparator`.
752 .. seealso::
754 :class:`.PropComparator`
756 :class:`.ColumnOperators`
758 :ref:`types_operators`
760 :attr:`.TypeEngine.comparator_factory`
762 """
764 # https://github.com/python/mypy/issues/4266
765 __hash__ = None # type: ignore
767 prop: RODescriptorReference[Composite[_PT]]
769 @util.memoized_property
770 def clauses(self) -> ClauseList:
771 return expression.ClauseList(
772 group=False, *self._comparable_elements
773 )
775 def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
776 return self.expression
778 @util.memoized_property
779 def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
780 clauses = self.clauses._annotate(
781 {
782 "parententity": self._parententity,
783 "parentmapper": self._parententity,
784 "proxy_key": self.prop.key,
785 }
786 )
787 return CompositeProperty.CompositeBundle(self.prop, clauses)
789 def _bulk_update_tuples(
790 self, value: Any
791 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
792 if isinstance(value, BindParameter):
793 value = value.value
795 values: Sequence[Any]
797 if value is None:
798 values = [None for key in self.prop._attribute_keys]
799 elif isinstance(self.prop.composite_class, type) and isinstance(
800 value, self.prop.composite_class
801 ):
802 values = self.prop._composite_values_from_instance(
803 value # type: ignore[arg-type]
804 )
805 else:
806 raise sa_exc.ArgumentError(
807 "Can't UPDATE composite attribute %s to %r"
808 % (self.prop, value)
809 )
811 return list(zip(self._comparable_elements, values))
813 @util.memoized_property
814 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
815 if self._adapt_to_entity:
816 return [
817 getattr(self._adapt_to_entity.entity, prop.key)
818 for prop in self.prop._comparable_elements
819 ]
820 else:
821 return self.prop._comparable_elements
823 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
824 return self._compare(operators.eq, other)
826 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
827 return self._compare(operators.ne, other)
829 def __lt__(self, other: Any) -> ColumnElement[bool]:
830 return self._compare(operators.lt, other)
832 def __gt__(self, other: Any) -> ColumnElement[bool]:
833 return self._compare(operators.gt, other)
835 def __le__(self, other: Any) -> ColumnElement[bool]:
836 return self._compare(operators.le, other)
838 def __ge__(self, other: Any) -> ColumnElement[bool]:
839 return self._compare(operators.ge, other)
841 # what might be interesting would be if we create
842 # an instance of the composite class itself with
843 # the columns as data members, then use "hybrid style" comparison
844 # to create these comparisons. then your Point.__eq__() method could
845 # be where comparison behavior is defined for SQL also. Likely
846 # not a good choice for default behavior though, not clear how it would
847 # work w/ dataclasses, etc. also no demand for any of this anyway.
848 def _compare(
849 self, operator: OperatorType, other: Any
850 ) -> ColumnElement[bool]:
851 values: Sequence[Any]
852 if other is None:
853 values = [None] * len(self.prop._comparable_elements)
854 else:
855 values = self.prop._composite_values_from_instance(other)
856 comparisons = [
857 operator(a, b)
858 for a, b in zip(self.prop._comparable_elements, values)
859 ]
860 if self._adapt_to_entity:
861 assert self.adapter is not None
862 comparisons = [self.adapter(x) for x in comparisons]
863 return sql.and_(*comparisons)
865 def __str__(self) -> str:
866 return str(self.parent.class_.__name__) + "." + self.key
869class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
870 """Declarative-compatible front-end for the :class:`.CompositeProperty`
871 class.
873 Public constructor is the :func:`_orm.composite` function.
875 .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
876 compatible subclass of :class:`_orm.CompositeProperty`.
878 .. seealso::
880 :ref:`mapper_composite`
882 """
884 inherit_cache = True
885 """:meta private:"""
888class ConcreteInheritedProperty(DescriptorProperty[_T]):
889 """A 'do nothing' :class:`.MapperProperty` that disables
890 an attribute on a concrete subclass that is only present
891 on the inherited mapper, not the concrete classes' mapper.
893 Cases where this occurs include:
895 * When the superclass mapper is mapped against a
896 "polymorphic union", which includes all attributes from
897 all subclasses.
898 * When a relationship() is configured on an inherited mapper,
899 but not on the subclass mapper. Concrete mappers require
900 that relationship() is configured explicitly on each
901 subclass.
903 """
905 def _comparator_factory(
906 self, mapper: Mapper[Any]
907 ) -> Type[PropComparator[_T]]:
908 comparator_callable = None
910 for m in self.parent.iterate_to_root():
911 p = m._props[self.key]
912 if getattr(p, "comparator_factory", None) is not None:
913 comparator_callable = p.comparator_factory
914 break
915 assert comparator_callable is not None
916 return comparator_callable(p, mapper) # type: ignore
918 def __init__(self) -> None:
919 super().__init__()
921 def warn() -> NoReturn:
922 raise AttributeError(
923 "Concrete %s does not implement "
924 "attribute %r at the instance level. Add "
925 "this property explicitly to %s."
926 % (self.parent, self.key, self.parent)
927 )
929 class NoninheritedConcreteProp:
930 def __set__(s: Any, obj: Any, value: Any) -> NoReturn:
931 warn()
933 def __delete__(s: Any, obj: Any) -> NoReturn:
934 warn()
936 def __get__(s: Any, obj: Any, owner: Any) -> Any:
937 if obj is None:
938 return self.descriptor
939 warn()
941 self.descriptor = NoninheritedConcreteProp()
944class SynonymProperty(DescriptorProperty[_T]):
945 """Denote an attribute name as a synonym to a mapped property,
946 in that the attribute will mirror the value and expression behavior
947 of another attribute.
949 :class:`.Synonym` is constructed using the :func:`_orm.synonym`
950 function.
952 .. seealso::
954 :ref:`synonyms` - Overview of synonyms
956 """
958 comparator_factory: Optional[Type[PropComparator[_T]]]
960 def __init__(
961 self,
962 name: str,
963 map_column: Optional[bool] = None,
964 descriptor: Optional[Any] = None,
965 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
966 attribute_options: Optional[_AttributeOptions] = None,
967 info: Optional[_InfoType] = None,
968 doc: Optional[str] = None,
969 ):
970 super().__init__(attribute_options=attribute_options)
972 self.name = name
973 self.map_column = map_column
974 self.descriptor = descriptor
975 self.comparator_factory = comparator_factory
976 if doc:
977 self.doc = doc
978 elif descriptor and descriptor.__doc__:
979 self.doc = descriptor.__doc__
980 else:
981 self.doc = None
982 if info:
983 self.info.update(info)
985 util.set_creation_order(self)
987 if not TYPE_CHECKING:
989 @property
990 def uses_objects(self) -> bool:
991 return getattr(self.parent.class_, self.name).impl.uses_objects
993 # TODO: when initialized, check _proxied_object,
994 # emit a warning if its not a column-based property
996 @util.memoized_property
997 def _proxied_object(
998 self,
999 ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]:
1000 attr = getattr(self.parent.class_, self.name)
1001 if not hasattr(attr, "property") or not isinstance(
1002 attr.property, MapperProperty
1003 ):
1004 # attribute is a non-MapperProprerty proxy such as
1005 # hybrid or association proxy
1006 if isinstance(attr, attributes.QueryableAttribute):
1007 return attr.comparator
1008 elif isinstance(attr, SQLORMOperations):
1009 # assocaition proxy comes here
1010 return attr
1012 raise sa_exc.InvalidRequestError(
1013 """synonym() attribute "%s.%s" only supports """
1014 """ORM mapped attributes, got %r"""
1015 % (self.parent.class_.__name__, self.name, attr)
1016 )
1017 return attr.property
1019 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
1020 return (getattr(self.parent.class_, self.name),)
1022 def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]:
1023 prop = self._proxied_object
1025 if isinstance(prop, MapperProperty):
1026 if self.comparator_factory:
1027 comp = self.comparator_factory(prop, mapper)
1028 else:
1029 comp = prop.comparator_factory(prop, mapper)
1030 return comp
1031 else:
1032 return prop
1034 def get_history(
1035 self,
1036 state: InstanceState[Any],
1037 dict_: _InstanceDict,
1038 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1039 ) -> History:
1040 attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name)
1041 return attr.impl.get_history(state, dict_, passive=passive)
1043 def _get_dataclass_setup_options(
1044 self,
1045 decl_scan: _ClassScanMapperConfig,
1046 key: str,
1047 dataclass_setup_arguments: _DataclassArguments,
1048 ) -> _AttributeOptions:
1049 dataclasses_default = self._attribute_options.dataclasses_default
1050 if (
1051 dataclasses_default is not _NoArg.NO_ARG
1052 and not callable(dataclasses_default)
1053 and not getattr(
1054 decl_scan.cls, "_sa_disable_descriptor_defaults", False
1055 )
1056 ):
1057 proxied = decl_scan.collected_attributes[self.name]
1058 proxied_default = proxied._attribute_options.dataclasses_default
1059 if proxied_default != dataclasses_default:
1060 raise sa_exc.ArgumentError(
1061 f"Synonym {key!r} default argument "
1062 f"{dataclasses_default!r} must match the dataclasses "
1063 f"default value of proxied object {self.name!r}, "
1064 f"""currently {
1065 repr(proxied_default)
1066 if proxied_default is not _NoArg.NO_ARG
1067 else 'not set'}"""
1068 )
1069 self._default_scalar_value = dataclasses_default
1070 return self._attribute_options._replace(
1071 dataclasses_default=DONT_SET
1072 )
1074 return self._attribute_options
1076 @util.preload_module("sqlalchemy.orm.properties")
1077 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
1078 properties = util.preloaded.orm_properties
1080 if self.map_column:
1081 # implement the 'map_column' option.
1082 if self.key not in parent.persist_selectable.c:
1083 raise sa_exc.ArgumentError(
1084 "Can't compile synonym '%s': no column on table "
1085 "'%s' named '%s'"
1086 % (
1087 self.name,
1088 parent.persist_selectable.description,
1089 self.key,
1090 )
1091 )
1092 elif (
1093 parent.persist_selectable.c[self.key]
1094 in parent._columntoproperty
1095 and parent._columntoproperty[
1096 parent.persist_selectable.c[self.key]
1097 ].key
1098 == self.name
1099 ):
1100 raise sa_exc.ArgumentError(
1101 "Can't call map_column=True for synonym %r=%r, "
1102 "a ColumnProperty already exists keyed to the name "
1103 "%r for column %r"
1104 % (self.key, self.name, self.name, self.key)
1105 )
1106 p: ColumnProperty[Any] = properties.ColumnProperty(
1107 parent.persist_selectable.c[self.key]
1108 )
1109 parent._configure_property(self.name, p, init=init, setparent=True)
1110 p._mapped_by_synonym = self.key
1112 self.parent = parent
1115class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
1116 """Declarative front-end for the :class:`.SynonymProperty` class.
1118 Public constructor is the :func:`_orm.synonym` function.
1120 .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
1121 compatible subclass for :class:`_orm.SynonymProperty`
1123 .. seealso::
1125 :ref:`synonyms` - Overview of synonyms
1127 """
1129 inherit_cache = True
1130 """:meta private:"""