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 get_args
24from typing import List
25from typing import NoReturn
26from typing import Optional
27from typing import Sequence
28from typing import Tuple
29from typing import Type
30from typing import TYPE_CHECKING
31from typing import TypeVar
32from typing import Union
33import weakref
35from . import attributes
36from . import util as orm_util
37from .base import _DeclarativeMapped
38from .base import DONT_SET
39from .base import LoaderCallableStatus
40from .base import Mapped
41from .base import PassiveFlag
42from .base import SQLORMOperations
43from .interfaces import _AttributeOptions
44from .interfaces import _IntrospectsAnnotations
45from .interfaces import _MapsColumns
46from .interfaces import MapperProperty
47from .interfaces import PropComparator
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 de_optionalize_union_types
59from ..util.typing import includes_none
60from ..util.typing import is_fwd_ref
61from ..util.typing import is_pep593
62from ..util.typing import is_union
63from ..util.typing import TupleAny
64from ..util.typing import Unpack
67if typing.TYPE_CHECKING:
68 from ._typing import _InstanceDict
69 from ._typing import _RegistryType
70 from .attributes import History
71 from .attributes import InstrumentedAttribute
72 from .attributes import QueryableAttribute
73 from .context import _ORMCompileState
74 from .decl_base import _ClassScanMapperConfig
75 from .interfaces import _DataclassArguments
76 from .mapper import Mapper
77 from .properties import ColumnProperty
78 from .properties import MappedColumn
79 from .state import InstanceState
80 from ..engine.base import Connection
81 from ..engine.row import Row
82 from ..sql._typing import _DMLColumnArgument
83 from ..sql._typing import _InfoType
84 from ..sql.elements import ClauseList
85 from ..sql.elements import ColumnElement
86 from ..sql.operators import OperatorType
87 from ..sql.schema import Column
88 from ..sql.selectable import Select
89 from ..util.typing import _AnnotationScanType
90 from ..util.typing import CallableReference
91 from ..util.typing import DescriptorReference
92 from ..util.typing import RODescriptorReference
94_T = TypeVar("_T", bound=Any)
95_PT = TypeVar("_PT", bound=Any)
98class DescriptorProperty(MapperProperty[_T]):
99 """:class:`.MapperProperty` which proxies access to a
100 user-defined descriptor."""
102 doc: Optional[str] = None
104 uses_objects = False
105 _links_to_entity = False
107 descriptor: DescriptorReference[Any]
109 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
110 raise NotImplementedError(
111 "This MapperProperty does not implement column loader strategies"
112 )
114 def get_history(
115 self,
116 state: InstanceState[Any],
117 dict_: _InstanceDict,
118 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
119 ) -> History:
120 raise NotImplementedError()
122 def instrument_class(self, mapper: Mapper[Any]) -> None:
123 prop = self
125 class _ProxyImpl(attributes._AttributeImpl):
126 accepts_scalar_loader = False
127 load_on_unexpire = True
128 collection = False
130 @property
131 def uses_objects(self) -> bool: # type: ignore
132 return prop.uses_objects
134 def __init__(self, key: str):
135 self.key = key
137 def get_history(
138 self,
139 state: InstanceState[Any],
140 dict_: _InstanceDict,
141 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
142 ) -> History:
143 return prop.get_history(state, dict_, passive)
145 if self.descriptor is None:
146 desc = getattr(mapper.class_, self.key, None)
147 if mapper._is_userland_descriptor(self.key, desc):
148 self.descriptor = desc
150 if self.descriptor is None:
152 def fset(obj: Any, value: Any) -> None:
153 setattr(obj, self.name, value)
155 def fdel(obj: Any) -> None:
156 delattr(obj, self.name)
158 def fget(obj: Any) -> Any:
159 return getattr(obj, self.name)
161 self.descriptor = property(fget=fget, fset=fset, fdel=fdel)
163 proxy_attr = attributes._create_proxied_attribute(self.descriptor)(
164 self.parent.class_,
165 self.key,
166 self.descriptor,
167 lambda: self._comparator_factory(mapper),
168 doc=self.doc,
169 original_property=self,
170 )
172 proxy_attr.impl = _ProxyImpl(self.key)
173 mapper.class_manager.instrument_attribute(self.key, proxy_attr)
176_CompositeAttrType = Union[
177 str,
178 "Column[_T]",
179 "MappedColumn[_T]",
180 "InstrumentedAttribute[_T]",
181 "Mapped[_T]",
182]
185_CC = TypeVar("_CC", bound=Any)
188_composite_getters: weakref.WeakKeyDictionary[
189 Type[Any], Callable[[Any], Tuple[Any, ...]]
190] = weakref.WeakKeyDictionary()
193class CompositeProperty(
194 _MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC]
195):
196 """Defines a "composite" mapped attribute, representing a collection
197 of columns as one attribute.
199 :class:`.CompositeProperty` is constructed using the :func:`.composite`
200 function.
202 .. seealso::
204 :ref:`mapper_composite`
206 """
208 composite_class: Union[Type[_CC], Callable[..., _CC]]
209 attrs: Tuple[_CompositeAttrType[Any], ...]
211 _generated_composite_accessor: CallableReference[
212 Optional[Callable[[_CC], Tuple[Any, ...]]]
213 ]
215 comparator_factory: Type[Comparator[_CC]]
217 def __init__(
218 self,
219 _class_or_attr: Union[
220 None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any]
221 ] = None,
222 *attrs: _CompositeAttrType[Any],
223 return_none_on: Union[
224 _NoArg, None, Callable[..., bool]
225 ] = _NoArg.NO_ARG,
226 attribute_options: Optional[_AttributeOptions] = None,
227 active_history: bool = False,
228 deferred: bool = False,
229 group: Optional[str] = None,
230 comparator_factory: Optional[Type[Comparator[_CC]]] = None,
231 info: Optional[_InfoType] = None,
232 **kwargs: Any,
233 ):
234 super().__init__(attribute_options=attribute_options)
236 if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
237 self.attrs = (_class_or_attr,) + attrs
238 # will initialize within declarative_scan
239 self.composite_class = None # type: ignore
240 else:
241 self.composite_class = _class_or_attr # type: ignore
242 self.attrs = attrs
244 self.return_none_on = return_none_on
245 self.active_history = active_history
246 self.deferred = deferred
247 self.group = group
248 self.comparator_factory = (
249 comparator_factory
250 if comparator_factory is not None
251 else self.__class__.Comparator
252 )
253 self._generated_composite_accessor = None
254 if info is not None:
255 self.info.update(info)
257 util.set_creation_order(self)
258 self._create_descriptor()
259 self._init_accessor()
261 @util.memoized_property
262 def _construct_composite(self) -> Callable[..., Any]:
263 return_none_on = self.return_none_on
264 if callable(return_none_on):
266 def construct(*args: Any) -> Any:
267 if return_none_on(*args):
268 return None
269 else:
270 return self.composite_class(*args)
272 return construct
273 else:
274 return self.composite_class
276 def instrument_class(self, mapper: Mapper[Any]) -> None:
277 super().instrument_class(mapper)
278 self._setup_event_handlers()
280 def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]:
281 if self._generated_composite_accessor:
282 return self._generated_composite_accessor(value)
283 else:
284 try:
285 accessor = value.__composite_values__
286 except AttributeError as ae:
287 raise sa_exc.InvalidRequestError(
288 f"Composite class {self.composite_class.__name__} is not "
289 f"a dataclass and does not define a __composite_values__()"
290 " method; can't get state"
291 ) from ae
292 else:
293 return accessor() # type: ignore
295 def do_init(self) -> None:
296 """Initialization which occurs after the :class:`.Composite`
297 has been associated with its parent mapper.
299 """
300 self._setup_arguments_on_columns()
302 _COMPOSITE_FGET = object()
304 def _create_descriptor(self) -> None:
305 """Create the Python descriptor that will serve as
306 the access point on instances of the mapped class.
308 """
310 def fget(instance: Any) -> Any:
311 dict_ = attributes.instance_dict(instance)
312 state = attributes.instance_state(instance)
314 if self.key not in dict_:
315 # key not present. Iterate through related
316 # attributes, retrieve their values. This
317 # ensures they all load.
318 values = [
319 getattr(instance, key) for key in self._attribute_keys
320 ]
322 if self.key not in dict_:
323 dict_[self.key] = self._construct_composite(*values)
324 state.manager.dispatch.refresh(
325 state, self._COMPOSITE_FGET, [self.key]
326 )
328 return dict_.get(self.key, None)
330 def fset(instance: Any, value: Any) -> None:
331 if value is LoaderCallableStatus.DONT_SET:
332 return
334 dict_ = attributes.instance_dict(instance)
335 state = attributes.instance_state(instance)
336 attr = state.manager[self.key]
338 if attr.dispatch._active_history:
339 previous = fget(instance)
340 else:
341 previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
343 for fn in attr.dispatch.set:
344 value = fn(state, value, previous, attr.impl)
345 dict_[self.key] = value
346 if value is None:
347 for key in self._attribute_keys:
348 setattr(instance, key, None)
349 else:
350 for key, value in zip(
351 self._attribute_keys,
352 self._composite_values_from_instance(value),
353 ):
354 setattr(instance, key, value)
356 def fdel(instance: Any) -> None:
357 state = attributes.instance_state(instance)
358 dict_ = attributes.instance_dict(instance)
359 attr = state.manager[self.key]
361 if attr.dispatch._active_history:
362 previous = fget(instance)
363 dict_.pop(self.key, None)
364 else:
365 previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
367 attr = state.manager[self.key]
368 attr.dispatch.remove(state, previous, attr.impl)
369 for key in self._attribute_keys:
370 setattr(instance, key, None)
372 self.descriptor = property(fget, fset, fdel)
374 @util.preload_module("sqlalchemy.orm.properties")
375 def declarative_scan(
376 self,
377 decl_scan: _ClassScanMapperConfig,
378 registry: _RegistryType,
379 cls: Type[Any],
380 originating_module: Optional[str],
381 key: str,
382 mapped_container: Optional[Type[Mapped[Any]]],
383 annotation: Optional[_AnnotationScanType],
384 extracted_mapped_annotation: Optional[_AnnotationScanType],
385 is_dataclass_field: bool,
386 ) -> None:
387 MappedColumn = util.preloaded.orm_properties.MappedColumn
388 if (
389 self.composite_class is None
390 and extracted_mapped_annotation is None
391 ):
392 self._raise_for_required(key, cls)
393 argument = extracted_mapped_annotation
395 if is_pep593(argument):
396 argument = get_args(argument)[0]
398 if argument and self.composite_class is None:
399 if isinstance(argument, str) or is_fwd_ref(
400 argument, check_generic=True
401 ):
402 if originating_module is None:
403 str_arg = (
404 argument.__forward_arg__
405 if hasattr(argument, "__forward_arg__")
406 else str(argument)
407 )
408 raise sa_exc.ArgumentError(
409 f"Can't use forward ref {argument} for composite "
410 f"class argument; set up the type as Mapped[{str_arg}]"
411 )
412 argument = de_stringify_annotation(
413 cls, argument, originating_module, include_generic=True
414 )
416 if is_union(argument) and includes_none(argument):
417 if self.return_none_on is _NoArg.NO_ARG:
418 self.return_none_on = lambda *args: all(
419 arg is None for arg in args
420 )
421 argument = de_optionalize_union_types(argument)
423 self.composite_class = argument
425 if is_dataclass(self.composite_class):
426 self._setup_for_dataclass(
427 decl_scan, registry, cls, originating_module, key
428 )
429 else:
430 for attr in self.attrs:
431 if (
432 isinstance(attr, (MappedColumn, schema.Column))
433 and attr.name is None
434 ):
435 raise sa_exc.ArgumentError(
436 "Composite class column arguments must be named "
437 "unless a dataclass is used"
438 )
439 self._init_accessor()
441 def _init_accessor(self) -> None:
442 if is_dataclass(self.composite_class) and not hasattr(
443 self.composite_class, "__composite_values__"
444 ):
445 insp = inspect.signature(self.composite_class)
446 getter = operator.attrgetter(
447 *[p.name for p in insp.parameters.values()]
448 )
449 if len(insp.parameters) == 1:
450 self._generated_composite_accessor = lambda obj: (getter(obj),)
451 else:
452 self._generated_composite_accessor = getter
454 if (
455 self.composite_class is not None
456 and isinstance(self.composite_class, type)
457 and self.composite_class not in _composite_getters
458 ):
459 if self._generated_composite_accessor is not None:
460 _composite_getters[self.composite_class] = (
461 self._generated_composite_accessor
462 )
463 elif hasattr(self.composite_class, "__composite_values__"):
464 _composite_getters[self.composite_class] = (
465 lambda obj: obj.__composite_values__()
466 )
468 @util.preload_module("sqlalchemy.orm.properties")
469 @util.preload_module("sqlalchemy.orm.decl_base")
470 def _setup_for_dataclass(
471 self,
472 decl_scan: _ClassScanMapperConfig,
473 registry: _RegistryType,
474 cls: Type[Any],
475 originating_module: Optional[str],
476 key: str,
477 ) -> None:
478 MappedColumn = util.preloaded.orm_properties.MappedColumn
480 decl_base = util.preloaded.orm_decl_base
482 insp = inspect.signature(self.composite_class)
483 for param, attr in itertools.zip_longest(
484 insp.parameters.values(), self.attrs
485 ):
486 if param is None:
487 raise sa_exc.ArgumentError(
488 f"number of composite attributes "
489 f"{len(self.attrs)} exceeds "
490 f"that of the number of attributes in class "
491 f"{self.composite_class.__name__} {len(insp.parameters)}"
492 )
493 if attr is None:
494 # fill in missing attr spots with empty MappedColumn
495 attr = MappedColumn()
496 self.attrs += (attr,)
498 if isinstance(attr, MappedColumn):
499 attr.declarative_scan_for_composite(
500 decl_scan,
501 registry,
502 cls,
503 originating_module,
504 key,
505 param.name,
506 param.annotation,
507 )
508 elif isinstance(attr, schema.Column):
509 decl_base._undefer_column_name(param.name, attr)
511 @util.memoized_property
512 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
513 return [getattr(self.parent.class_, prop.key) for prop in self.props]
515 @util.memoized_property
516 @util.preload_module("orm.properties")
517 def props(self) -> Sequence[MapperProperty[Any]]:
518 props = []
519 MappedColumn = util.preloaded.orm_properties.MappedColumn
521 for attr in self.attrs:
522 if isinstance(attr, str):
523 prop = self.parent.get_property(attr, _configure_mappers=False)
524 elif isinstance(attr, schema.Column):
525 prop = self.parent._columntoproperty[attr]
526 elif isinstance(attr, MappedColumn):
527 prop = self.parent._columntoproperty[attr.column]
528 elif isinstance(attr, attributes.InstrumentedAttribute):
529 prop = attr.property
530 else:
531 prop = None
533 if not isinstance(prop, MapperProperty):
534 raise sa_exc.ArgumentError(
535 "Composite expects Column objects or mapped "
536 f"attributes/attribute names as arguments, got: {attr!r}"
537 )
539 props.append(prop)
540 return props
542 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
543 return self._comparable_elements
545 @util.non_memoized_property
546 @util.preload_module("orm.properties")
547 def columns(self) -> Sequence[Column[Any]]:
548 MappedColumn = util.preloaded.orm_properties.MappedColumn
549 return [
550 a.column if isinstance(a, MappedColumn) else a
551 for a in self.attrs
552 if isinstance(a, (schema.Column, MappedColumn))
553 ]
555 @property
556 def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]:
557 return self
559 @property
560 def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
561 return [(c, 0) for c in self.columns if c.table is None]
563 @util.preload_module("orm.properties")
564 def _setup_arguments_on_columns(self) -> None:
565 """Propagate configuration arguments made on this composite
566 to the target columns, for those that apply.
568 """
569 ColumnProperty = util.preloaded.orm_properties.ColumnProperty
571 for prop in self.props:
572 if not isinstance(prop, ColumnProperty):
573 continue
574 else:
575 cprop = prop
577 cprop.active_history = self.active_history
578 if self.deferred:
579 cprop.deferred = self.deferred
580 cprop.strategy_key = (("deferred", True), ("instrument", True))
581 cprop.group = self.group
583 def _setup_event_handlers(self) -> None:
584 """Establish events that populate/expire the composite attribute."""
586 def load_handler(
587 state: InstanceState[Any], context: _ORMCompileState
588 ) -> None:
589 _load_refresh_handler(state, context, None, is_refresh=False)
591 def refresh_handler(
592 state: InstanceState[Any],
593 context: _ORMCompileState,
594 to_load: Optional[Sequence[str]],
595 ) -> None:
596 # note this corresponds to sqlalchemy.ext.mutable load_attrs()
598 if not to_load or (
599 {self.key}.union(self._attribute_keys)
600 ).intersection(to_load):
601 _load_refresh_handler(state, context, to_load, is_refresh=True)
603 def _load_refresh_handler(
604 state: InstanceState[Any],
605 context: _ORMCompileState,
606 to_load: Optional[Sequence[str]],
607 is_refresh: bool,
608 ) -> None:
609 dict_ = state.dict
611 # if context indicates we are coming from the
612 # fget() handler, this already set the value; skip the
613 # handler here. (other handlers like mutablecomposite will still
614 # want to catch it)
615 # there's an insufficiency here in that the fget() handler
616 # really should not be using the refresh event and there should
617 # be some other event that mutablecomposite can subscribe
618 # towards for this.
620 if (
621 not is_refresh or context is self._COMPOSITE_FGET
622 ) and self.key in dict_:
623 return
625 # if column elements aren't loaded, skip.
626 # __get__() will initiate a load for those
627 # columns
628 for k in self._attribute_keys:
629 if k not in dict_:
630 return
632 dict_[self.key] = self._construct_composite(
633 *[state.dict[key] for key in self._attribute_keys]
634 )
636 def expire_handler(
637 state: InstanceState[Any], keys: Optional[Sequence[str]]
638 ) -> None:
639 if keys is None or set(self._attribute_keys).intersection(keys):
640 state.dict.pop(self.key, None)
642 def insert_update_handler(
643 mapper: Mapper[Any],
644 connection: Connection,
645 state: InstanceState[Any],
646 ) -> None:
647 """After an insert or update, some columns may be expired due
648 to server side defaults, or re-populated due to client side
649 defaults. Pop out the composite value here so that it
650 recreates.
652 """
654 state.dict.pop(self.key, None)
656 event.listen(
657 self.parent, "after_insert", insert_update_handler, raw=True
658 )
659 event.listen(
660 self.parent, "after_update", insert_update_handler, raw=True
661 )
662 event.listen(
663 self.parent, "load", load_handler, raw=True, propagate=True
664 )
665 event.listen(
666 self.parent, "refresh", refresh_handler, raw=True, propagate=True
667 )
668 event.listen(
669 self.parent, "expire", expire_handler, raw=True, propagate=True
670 )
672 proxy_attr = self.parent.class_manager[self.key]
673 proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore
674 proxy_attr.impl.dispatch._active_history = self.active_history
676 # TODO: need a deserialize hook here
678 @util.memoized_property
679 def _attribute_keys(self) -> Sequence[str]:
680 return [prop.key for prop in self.props]
682 def _populate_composite_bulk_save_mappings_fn(
683 self,
684 ) -> Callable[[Dict[str, Any]], None]:
685 if self._generated_composite_accessor:
686 get_values = self._generated_composite_accessor
687 else:
689 def get_values(val: Any) -> Tuple[Any]:
690 return val.__composite_values__() # type: ignore
692 attrs = [prop.key for prop in self.props]
694 def populate(dest_dict: Dict[str, Any]) -> None:
695 dest_dict.update(
696 {
697 key: val
698 for key, val in zip(
699 attrs, get_values(dest_dict.pop(self.key))
700 )
701 }
702 )
704 return populate
706 def get_history(
707 self,
708 state: InstanceState[Any],
709 dict_: _InstanceDict,
710 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
711 ) -> History:
712 """Provided for userland code that uses attributes.get_history()."""
714 added: List[Any] = []
715 deleted: List[Any] = []
717 has_history = False
718 for prop in self.props:
719 key = prop.key
720 hist = state.manager[key].impl.get_history(state, dict_)
721 if hist.has_changes():
722 has_history = True
724 non_deleted = hist.non_deleted()
725 if non_deleted:
726 added.extend(non_deleted)
727 else:
728 added.append(None)
729 if hist.deleted:
730 deleted.extend(hist.deleted)
731 else:
732 deleted.append(None)
734 if has_history:
735 return attributes.History(
736 [self._construct_composite(*added)],
737 (),
738 [self._construct_composite(*deleted)],
739 )
740 else:
741 return attributes.History(
742 (), [self._construct_composite(*added)], ()
743 )
745 def _comparator_factory(
746 self, mapper: Mapper[Any]
747 ) -> Composite.Comparator[_CC]:
748 return self.comparator_factory(self, mapper)
750 class CompositeBundle(orm_util.Bundle[_T]):
751 def __init__(
752 self,
753 property_: Composite[_T],
754 expr: ClauseList,
755 ):
756 self.property = property_
757 super().__init__(property_.key, *expr)
759 def create_row_processor(
760 self,
761 query: Select[Unpack[TupleAny]],
762 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
763 labels: Sequence[str],
764 ) -> Callable[[Row[Unpack[TupleAny]]], Any]:
765 def proc(row: Row[Unpack[TupleAny]]) -> Any:
766 return self.property._construct_composite(
767 *[proc(row) for proc in procs]
768 )
770 return proc
772 class Comparator(PropComparator[_PT]):
773 """Produce boolean, comparison, and other operators for
774 :class:`.Composite` attributes.
776 See the example in :ref:`composite_operations` for an overview
777 of usage , as well as the documentation for :class:`.PropComparator`.
779 .. seealso::
781 :class:`.PropComparator`
783 :class:`.ColumnOperators`
785 :ref:`types_operators`
787 :attr:`.TypeEngine.comparator_factory`
789 """
791 # https://github.com/python/mypy/issues/4266
792 __hash__ = None # type: ignore
794 prop: RODescriptorReference[Composite[_PT]]
796 @util.memoized_property
797 def clauses(self) -> ClauseList:
798 return expression.ClauseList(
799 group=False, *self._comparable_elements
800 )
802 def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
803 return self.expression
805 @util.memoized_property
806 def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
807 clauses = self.clauses._annotate(
808 {
809 "parententity": self._parententity,
810 "parentmapper": self._parententity,
811 "proxy_key": self.prop.key,
812 }
813 )
814 return CompositeProperty.CompositeBundle(self.prop, clauses)
816 def _bulk_update_tuples(
817 self, value: Any
818 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
819 if isinstance(value, BindParameter):
820 value = value.value
822 values: Sequence[Any]
824 if value is None:
825 values = [None for key in self.prop._attribute_keys]
826 elif isinstance(self.prop.composite_class, type) and isinstance(
827 value, self.prop.composite_class
828 ):
829 values = self.prop._composite_values_from_instance(
830 value # type: ignore[arg-type]
831 )
832 else:
833 raise sa_exc.ArgumentError(
834 "Can't UPDATE composite attribute %s to %r"
835 % (self.prop, value)
836 )
838 return list(zip(self._comparable_elements, values))
840 def _bulk_dml_setter(self, key: str) -> Optional[Callable[..., Any]]:
841 return self.prop._populate_composite_bulk_save_mappings_fn()
843 @util.memoized_property
844 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
845 if self._adapt_to_entity:
846 return [
847 getattr(self._adapt_to_entity.entity, prop.key)
848 for prop in self.prop._comparable_elements
849 ]
850 else:
851 return self.prop._comparable_elements
853 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
854 return self._compare(operators.eq, other)
856 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
857 return self._compare(operators.ne, other)
859 def __lt__(self, other: Any) -> ColumnElement[bool]:
860 return self._compare(operators.lt, other)
862 def __gt__(self, other: Any) -> ColumnElement[bool]:
863 return self._compare(operators.gt, other)
865 def __le__(self, other: Any) -> ColumnElement[bool]:
866 return self._compare(operators.le, other)
868 def __ge__(self, other: Any) -> ColumnElement[bool]:
869 return self._compare(operators.ge, other)
871 def desc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
872 return expression.OrderByList(
873 [e.desc() for e in self._comparable_elements]
874 )
876 def asc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
877 return expression.OrderByList(
878 [e.asc() for e in self._comparable_elements]
879 )
881 def nulls_first(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
882 return expression.OrderByList(
883 [e.nulls_first() for e in self._comparable_elements]
884 )
886 def nulls_last(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
887 return expression.OrderByList(
888 [e.nulls_last() for e in self._comparable_elements]
889 )
891 # what might be interesting would be if we create
892 # an instance of the composite class itself with
893 # the columns as data members, then use "hybrid style" comparison
894 # to create these comparisons. then your Point.__eq__() method could
895 # be where comparison behavior is defined for SQL also. Likely
896 # not a good choice for default behavior though, not clear how it would
897 # work w/ dataclasses, etc. also no demand for any of this anyway.
898 def _compare(
899 self, operator: OperatorType, other: Any
900 ) -> ColumnElement[bool]:
901 values: Sequence[Any]
902 if other is None:
903 values = [None] * len(self.prop._comparable_elements)
904 else:
905 values = self.prop._composite_values_from_instance(other)
906 comparisons = [
907 operator(a, b)
908 for a, b in zip(self.prop._comparable_elements, values)
909 ]
910 if self._adapt_to_entity:
911 assert self.adapter is not None
912 comparisons = [self.adapter(x) for x in comparisons]
913 return sql.and_(*comparisons)
915 def __str__(self) -> str:
916 return str(self.parent.class_.__name__) + "." + self.key
919class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
920 """Declarative-compatible front-end for the :class:`.CompositeProperty`
921 class.
923 Public constructor is the :func:`_orm.composite` function.
925 .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
926 compatible subclass of :class:`_orm.CompositeProperty`.
928 .. seealso::
930 :ref:`mapper_composite`
932 """
934 inherit_cache = True
935 """:meta private:"""
938class ConcreteInheritedProperty(DescriptorProperty[_T]):
939 """A 'do nothing' :class:`.MapperProperty` that disables
940 an attribute on a concrete subclass that is only present
941 on the inherited mapper, not the concrete classes' mapper.
943 Cases where this occurs include:
945 * When the superclass mapper is mapped against a
946 "polymorphic union", which includes all attributes from
947 all subclasses.
948 * When a relationship() is configured on an inherited mapper,
949 but not on the subclass mapper. Concrete mappers require
950 that relationship() is configured explicitly on each
951 subclass.
953 """
955 def _comparator_factory(
956 self, mapper: Mapper[Any]
957 ) -> Type[PropComparator[_T]]:
958 comparator_callable = None
960 for m in self.parent.iterate_to_root():
961 p = m._props[self.key]
962 if getattr(p, "comparator_factory", None) is not None:
963 comparator_callable = p.comparator_factory
964 break
965 assert comparator_callable is not None
966 return comparator_callable(p, mapper) # type: ignore
968 def __init__(self) -> None:
969 super().__init__()
971 def warn() -> NoReturn:
972 raise AttributeError(
973 "Concrete %s does not implement "
974 "attribute %r at the instance level. Add "
975 "this property explicitly to %s."
976 % (self.parent, self.key, self.parent)
977 )
979 class NoninheritedConcreteProp:
980 def __set__(s: Any, obj: Any, value: Any) -> NoReturn:
981 warn()
983 def __delete__(s: Any, obj: Any) -> NoReturn:
984 warn()
986 def __get__(s: Any, obj: Any, owner: Any) -> Any:
987 if obj is None:
988 return self.descriptor
989 warn()
991 self.descriptor = NoninheritedConcreteProp()
994class SynonymProperty(DescriptorProperty[_T]):
995 """Denote an attribute name as a synonym to a mapped property,
996 in that the attribute will mirror the value and expression behavior
997 of another attribute.
999 :class:`.Synonym` is constructed using the :func:`_orm.synonym`
1000 function.
1002 .. seealso::
1004 :ref:`synonyms` - Overview of synonyms
1006 """
1008 comparator_factory: Optional[Type[PropComparator[_T]]]
1010 def __init__(
1011 self,
1012 name: str,
1013 map_column: Optional[bool] = None,
1014 descriptor: Optional[Any] = None,
1015 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
1016 attribute_options: Optional[_AttributeOptions] = None,
1017 info: Optional[_InfoType] = None,
1018 doc: Optional[str] = None,
1019 ):
1020 super().__init__(attribute_options=attribute_options)
1022 self.name = name
1023 self.map_column = map_column
1024 self.descriptor = descriptor
1025 self.comparator_factory = comparator_factory
1026 if doc:
1027 self.doc = doc
1028 elif descriptor and descriptor.__doc__:
1029 self.doc = descriptor.__doc__
1030 else:
1031 self.doc = None
1032 if info:
1033 self.info.update(info)
1035 util.set_creation_order(self)
1037 if not TYPE_CHECKING:
1039 @property
1040 def uses_objects(self) -> bool:
1041 return getattr(self.parent.class_, self.name).impl.uses_objects
1043 # TODO: when initialized, check _proxied_object,
1044 # emit a warning if its not a column-based property
1046 @util.memoized_property
1047 def _proxied_object(
1048 self,
1049 ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]:
1050 attr = getattr(self.parent.class_, self.name)
1051 if not hasattr(attr, "property") or not isinstance(
1052 attr.property, MapperProperty
1053 ):
1054 # attribute is a non-MapperProprerty proxy such as
1055 # hybrid or association proxy
1056 if isinstance(attr, attributes.QueryableAttribute):
1057 return attr.comparator
1058 elif isinstance(attr, SQLORMOperations):
1059 # assocaition proxy comes here
1060 return attr
1062 raise sa_exc.InvalidRequestError(
1063 """synonym() attribute "%s.%s" only supports """
1064 """ORM mapped attributes, got %r"""
1065 % (self.parent.class_.__name__, self.name, attr)
1066 )
1067 return attr.property
1069 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
1070 return (getattr(self.parent.class_, self.name),)
1072 def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]:
1073 prop = self._proxied_object
1075 if isinstance(prop, MapperProperty):
1076 if self.comparator_factory:
1077 comp = self.comparator_factory(prop, mapper)
1078 else:
1079 comp = prop.comparator_factory(prop, mapper)
1080 return comp
1081 else:
1082 return prop
1084 def get_history(
1085 self,
1086 state: InstanceState[Any],
1087 dict_: _InstanceDict,
1088 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1089 ) -> History:
1090 attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name)
1091 return attr.impl.get_history(state, dict_, passive=passive)
1093 def _get_dataclass_setup_options(
1094 self,
1095 decl_scan: _ClassScanMapperConfig,
1096 key: str,
1097 dataclass_setup_arguments: _DataclassArguments,
1098 ) -> _AttributeOptions:
1099 dataclasses_default = self._attribute_options.dataclasses_default
1100 if (
1101 dataclasses_default is not _NoArg.NO_ARG
1102 and not callable(dataclasses_default)
1103 and not getattr(
1104 decl_scan.cls, "_sa_disable_descriptor_defaults", False
1105 )
1106 ):
1107 proxied = decl_scan.collected_attributes[self.name]
1108 proxied_default = proxied._attribute_options.dataclasses_default
1109 if proxied_default != dataclasses_default:
1110 raise sa_exc.ArgumentError(
1111 f"Synonym {key!r} default argument "
1112 f"{dataclasses_default!r} must match the dataclasses "
1113 f"default value of proxied object {self.name!r}, "
1114 f"""currently {
1115 repr(proxied_default)
1116 if proxied_default is not _NoArg.NO_ARG
1117 else 'not set'}"""
1118 )
1119 self._default_scalar_value = dataclasses_default
1120 return self._attribute_options._replace(
1121 dataclasses_default=DONT_SET
1122 )
1124 return self._attribute_options
1126 @util.preload_module("sqlalchemy.orm.properties")
1127 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
1128 properties = util.preloaded.orm_properties
1130 if self.map_column:
1131 # implement the 'map_column' option.
1132 if self.key not in parent.persist_selectable.c:
1133 raise sa_exc.ArgumentError(
1134 "Can't compile synonym '%s': no column on table "
1135 "'%s' named '%s'"
1136 % (
1137 self.name,
1138 parent.persist_selectable.description,
1139 self.key,
1140 )
1141 )
1142 elif (
1143 parent.persist_selectable.c[self.key]
1144 in parent._columntoproperty
1145 and parent._columntoproperty[
1146 parent.persist_selectable.c[self.key]
1147 ].key
1148 == self.name
1149 ):
1150 raise sa_exc.ArgumentError(
1151 "Can't call map_column=True for synonym %r=%r, "
1152 "a ColumnProperty already exists keyed to the name "
1153 "%r for column %r"
1154 % (self.key, self.name, self.name, self.key)
1155 )
1156 p: ColumnProperty[Any] = properties.ColumnProperty(
1157 parent.persist_selectable.c[self.key]
1158 )
1159 parent._configure_property(self.name, p, init=init, setparent=True)
1160 p._mapped_by_synonym = self.key
1162 self.parent = parent
1165class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
1166 """Declarative front-end for the :class:`.SynonymProperty` class.
1168 Public constructor is the :func:`_orm.synonym` function.
1170 .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
1171 compatible subclass for :class:`_orm.SynonymProperty`
1173 .. seealso::
1175 :ref:`synonyms` - Overview of synonyms
1177 """
1179 inherit_cache = True
1180 """:meta private:"""