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 _ClassScanAbstractConfig
75 from .decl_base import _DeclarativeMapperConfig
76 from .interfaces import _DataclassArguments
77 from .mapper import Mapper
78 from .properties import ColumnProperty
79 from .properties import MappedColumn
80 from .state import InstanceState
81 from ..engine.base import Connection
82 from ..engine.row import Row
83 from ..sql._typing import _DMLColumnArgument
84 from ..sql._typing import _InfoType
85 from ..sql.elements import ClauseList
86 from ..sql.elements import ColumnElement
87 from ..sql.operators import OperatorType
88 from ..sql.schema import Column
89 from ..sql.selectable import Select
90 from ..util.typing import _AnnotationScanType
91 from ..util.typing import CallableReference
92 from ..util.typing import DescriptorReference
93 from ..util.typing import RODescriptorReference
95_T = TypeVar("_T", bound=Any)
96_PT = TypeVar("_PT", bound=Any)
99class DescriptorProperty(MapperProperty[_T]):
100 """:class:`.MapperProperty` which proxies access to a
101 user-defined descriptor."""
103 doc: Optional[str] = None
105 uses_objects = False
106 _links_to_entity = False
108 descriptor: DescriptorReference[Any]
110 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
111 raise NotImplementedError(
112 "This MapperProperty does not implement column loader strategies"
113 )
115 def get_history(
116 self,
117 state: InstanceState[Any],
118 dict_: _InstanceDict,
119 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
120 ) -> History:
121 raise NotImplementedError()
123 def instrument_class(self, mapper: Mapper[Any]) -> None:
124 prop = self
126 class _ProxyImpl(attributes._AttributeImpl):
127 accepts_scalar_loader = False
128 load_on_unexpire = True
129 collection = False
131 @property
132 def uses_objects(self) -> bool: # type: ignore
133 return prop.uses_objects
135 def __init__(self, key: str):
136 self.key = key
138 def get_history(
139 self,
140 state: InstanceState[Any],
141 dict_: _InstanceDict,
142 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
143 ) -> History:
144 return prop.get_history(state, dict_, passive)
146 if self.descriptor is None:
147 desc = getattr(mapper.class_, self.key, None)
148 if mapper._is_userland_descriptor(self.key, desc):
149 self.descriptor = desc
151 if self.descriptor is None:
153 def fset(obj: Any, value: Any) -> None:
154 setattr(obj, self.name, value)
156 def fdel(obj: Any) -> None:
157 delattr(obj, self.name)
159 def fget(obj: Any) -> Any:
160 return getattr(obj, self.name)
162 self.descriptor = property(fget=fget, fset=fset, fdel=fdel)
164 proxy_attr = attributes._create_proxied_attribute(self.descriptor)(
165 self.parent.class_,
166 self.key,
167 self.descriptor,
168 lambda: self._comparator_factory(mapper),
169 doc=self.doc,
170 original_property=self,
171 )
173 proxy_attr.impl = _ProxyImpl(self.key)
174 mapper.class_manager.instrument_attribute(self.key, proxy_attr)
177_CompositeAttrType = Union[
178 str,
179 "Column[_T]",
180 "MappedColumn[_T]",
181 "InstrumentedAttribute[_T]",
182 "Mapped[_T]",
183]
186_CC = TypeVar("_CC", bound=Any)
189_composite_getters: weakref.WeakKeyDictionary[
190 Type[Any], Callable[[Any], Tuple[Any, ...]]
191] = weakref.WeakKeyDictionary()
194class CompositeProperty(
195 _MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC]
196):
197 """Defines a "composite" mapped attribute, representing a collection
198 of columns as one attribute.
200 :class:`.CompositeProperty` is constructed using the :func:`.composite`
201 function.
203 .. seealso::
205 :ref:`mapper_composite`
207 """
209 composite_class: Union[Type[_CC], Callable[..., _CC]]
210 attrs: Tuple[_CompositeAttrType[Any], ...]
212 _generated_composite_accessor: CallableReference[
213 Optional[Callable[[_CC], Tuple[Any, ...]]]
214 ]
216 comparator_factory: Type[Comparator[_CC]]
218 def __init__(
219 self,
220 _class_or_attr: Union[
221 None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any]
222 ] = None,
223 *attrs: _CompositeAttrType[Any],
224 return_none_on: Union[
225 _NoArg, None, Callable[..., bool]
226 ] = _NoArg.NO_ARG,
227 attribute_options: Optional[_AttributeOptions] = None,
228 active_history: bool = False,
229 deferred: bool = False,
230 group: Optional[str] = None,
231 comparator_factory: Optional[Type[Comparator[_CC]]] = None,
232 info: Optional[_InfoType] = None,
233 **kwargs: Any,
234 ):
235 super().__init__(attribute_options=attribute_options)
237 if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
238 self.attrs = (_class_or_attr,) + attrs
239 # will initialize within declarative_scan
240 self.composite_class = None # type: ignore
241 else:
242 self.composite_class = _class_or_attr # type: ignore
243 self.attrs = attrs
245 self.return_none_on = return_none_on
246 self.active_history = active_history
247 self.deferred = deferred
248 self.group = group
249 self.comparator_factory = (
250 comparator_factory
251 if comparator_factory is not None
252 else self.__class__.Comparator
253 )
254 self._generated_composite_accessor = None
255 if info is not None:
256 self.info.update(info)
258 util.set_creation_order(self)
259 self._create_descriptor()
260 self._init_accessor()
262 @util.memoized_property
263 def _construct_composite(self) -> Callable[..., Any]:
264 return_none_on = self.return_none_on
265 if callable(return_none_on):
267 def construct(*args: Any) -> Any:
268 if return_none_on(*args):
269 return None
270 else:
271 return self.composite_class(*args)
273 return construct
274 else:
275 return self.composite_class
277 def instrument_class(self, mapper: Mapper[Any]) -> None:
278 super().instrument_class(mapper)
279 self._setup_event_handlers()
281 def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]:
282 if self._generated_composite_accessor:
283 return self._generated_composite_accessor(value)
284 else:
285 try:
286 accessor = value.__composite_values__
287 except AttributeError as ae:
288 raise sa_exc.InvalidRequestError(
289 f"Composite class {self.composite_class.__name__} is not "
290 f"a dataclass and does not define a __composite_values__()"
291 " method; can't get state"
292 ) from ae
293 else:
294 return accessor() # type: ignore
296 def do_init(self) -> None:
297 """Initialization which occurs after the :class:`.Composite`
298 has been associated with its parent mapper.
300 """
301 self._setup_arguments_on_columns()
303 _COMPOSITE_FGET = object()
305 def _create_descriptor(self) -> None:
306 """Create the Python descriptor that will serve as
307 the access point on instances of the mapped class.
309 """
311 def fget(instance: Any) -> Any:
312 dict_ = attributes.instance_dict(instance)
313 state = attributes.instance_state(instance)
315 if self.key not in dict_:
316 # key not present. Iterate through related
317 # attributes, retrieve their values. This
318 # ensures they all load.
319 values = [
320 getattr(instance, key) for key in self._attribute_keys
321 ]
323 if self.key not in dict_:
324 dict_[self.key] = self._construct_composite(*values)
325 state.manager.dispatch.refresh(
326 state, self._COMPOSITE_FGET, [self.key]
327 )
329 return dict_.get(self.key, None)
331 def fset(instance: Any, value: Any) -> None:
332 if value is LoaderCallableStatus.DONT_SET:
333 return
335 dict_ = attributes.instance_dict(instance)
336 state = attributes.instance_state(instance)
337 attr = state.manager[self.key]
339 if attr.dispatch._active_history:
340 previous = fget(instance)
341 else:
342 previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
344 for fn in attr.dispatch.set:
345 value = fn(state, value, previous, attr.impl)
346 dict_[self.key] = value
347 if value is None:
348 for key in self._attribute_keys:
349 setattr(instance, key, None)
350 else:
351 for key, value in zip(
352 self._attribute_keys,
353 self._composite_values_from_instance(value),
354 ):
355 setattr(instance, key, value)
357 def fdel(instance: Any) -> None:
358 state = attributes.instance_state(instance)
359 dict_ = attributes.instance_dict(instance)
360 attr = state.manager[self.key]
362 if attr.dispatch._active_history:
363 previous = fget(instance)
364 dict_.pop(self.key, None)
365 else:
366 previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
368 attr = state.manager[self.key]
369 attr.dispatch.remove(state, previous, attr.impl)
370 for key in self._attribute_keys:
371 setattr(instance, key, None)
373 self.descriptor = property(fget, fset, fdel)
375 @util.preload_module("sqlalchemy.orm.properties")
376 def declarative_scan(
377 self,
378 decl_scan: _DeclarativeMapperConfig,
379 registry: _RegistryType,
380 cls: Type[Any],
381 originating_module: Optional[str],
382 key: str,
383 mapped_container: Optional[Type[Mapped[Any]]],
384 annotation: Optional[_AnnotationScanType],
385 extracted_mapped_annotation: Optional[_AnnotationScanType],
386 is_dataclass_field: bool,
387 ) -> None:
388 MappedColumn = util.preloaded.orm_properties.MappedColumn
389 if (
390 self.composite_class is None
391 and extracted_mapped_annotation is None
392 ):
393 self._raise_for_required(key, cls)
394 argument = extracted_mapped_annotation
396 if is_pep593(argument):
397 argument = get_args(argument)[0]
399 if argument and self.composite_class is None:
400 if isinstance(argument, str) or is_fwd_ref(
401 argument, check_generic=True
402 ):
403 if originating_module is None:
404 str_arg = (
405 argument.__forward_arg__
406 if hasattr(argument, "__forward_arg__")
407 else str(argument)
408 )
409 raise sa_exc.ArgumentError(
410 f"Can't use forward ref {argument} for composite "
411 f"class argument; set up the type as Mapped[{str_arg}]"
412 )
413 argument = de_stringify_annotation(
414 cls, argument, originating_module, include_generic=True
415 )
417 if is_union(argument) and includes_none(argument):
418 if self.return_none_on is _NoArg.NO_ARG:
419 self.return_none_on = lambda *args: all(
420 arg is None for arg in args
421 )
422 argument = de_optionalize_union_types(argument)
424 self.composite_class = argument
426 if is_dataclass(self.composite_class):
427 self._setup_for_dataclass(
428 decl_scan, registry, cls, originating_module, key
429 )
430 else:
431 for attr in self.attrs:
432 if (
433 isinstance(attr, (MappedColumn, schema.Column))
434 and attr.name is None
435 ):
436 raise sa_exc.ArgumentError(
437 "Composite class column arguments must be named "
438 "unless a dataclass is used"
439 )
440 self._init_accessor()
442 def _init_accessor(self) -> None:
443 if is_dataclass(self.composite_class) and not hasattr(
444 self.composite_class, "__composite_values__"
445 ):
446 insp = inspect.signature(self.composite_class)
447 getter = operator.attrgetter(
448 *[p.name for p in insp.parameters.values()]
449 )
450 if len(insp.parameters) == 1:
451 self._generated_composite_accessor = lambda obj: (getter(obj),)
452 else:
453 self._generated_composite_accessor = getter
455 if (
456 self.composite_class is not None
457 and isinstance(self.composite_class, type)
458 and self.composite_class not in _composite_getters
459 ):
460 if self._generated_composite_accessor is not None:
461 _composite_getters[self.composite_class] = (
462 self._generated_composite_accessor
463 )
464 elif hasattr(self.composite_class, "__composite_values__"):
465 _composite_getters[self.composite_class] = (
466 lambda obj: obj.__composite_values__()
467 )
469 @util.preload_module("sqlalchemy.orm.properties")
470 @util.preload_module("sqlalchemy.orm.decl_base")
471 def _setup_for_dataclass(
472 self,
473 decl_scan: _DeclarativeMapperConfig,
474 registry: _RegistryType,
475 cls: Type[Any],
476 originating_module: Optional[str],
477 key: str,
478 ) -> None:
479 MappedColumn = util.preloaded.orm_properties.MappedColumn
481 decl_base = util.preloaded.orm_decl_base
483 insp = inspect.signature(self.composite_class)
484 for param, attr in itertools.zip_longest(
485 insp.parameters.values(), self.attrs
486 ):
487 if param is None:
488 raise sa_exc.ArgumentError(
489 f"number of composite attributes "
490 f"{len(self.attrs)} exceeds "
491 f"that of the number of attributes in class "
492 f"{self.composite_class.__name__} {len(insp.parameters)}"
493 )
494 if attr is None:
495 # fill in missing attr spots with empty MappedColumn
496 attr = MappedColumn()
497 self.attrs += (attr,)
499 if isinstance(attr, MappedColumn):
500 attr.declarative_scan_for_composite(
501 decl_scan,
502 registry,
503 cls,
504 originating_module,
505 key,
506 param.name,
507 param.annotation,
508 )
509 elif isinstance(attr, schema.Column):
510 decl_base._undefer_column_name(param.name, attr)
512 @util.memoized_property
513 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
514 return [getattr(self.parent.class_, prop.key) for prop in self.props]
516 @util.memoized_property
517 @util.preload_module("orm.properties")
518 def props(self) -> Sequence[MapperProperty[Any]]:
519 props = []
520 MappedColumn = util.preloaded.orm_properties.MappedColumn
522 for attr in self.attrs:
523 if isinstance(attr, str):
524 prop = self.parent.get_property(attr, _configure_mappers=False)
525 elif isinstance(attr, schema.Column):
526 prop = self.parent._columntoproperty[attr]
527 elif isinstance(attr, MappedColumn):
528 prop = self.parent._columntoproperty[attr.column]
529 elif isinstance(attr, attributes.InstrumentedAttribute):
530 prop = attr.property
531 else:
532 prop = None
534 if not isinstance(prop, MapperProperty):
535 raise sa_exc.ArgumentError(
536 "Composite expects Column objects or mapped "
537 f"attributes/attribute names as arguments, got: {attr!r}"
538 )
540 props.append(prop)
541 return props
543 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
544 return self._comparable_elements
546 @util.non_memoized_property
547 @util.preload_module("orm.properties")
548 def columns(self) -> Sequence[Column[Any]]:
549 MappedColumn = util.preloaded.orm_properties.MappedColumn
550 return [
551 a.column if isinstance(a, MappedColumn) else a
552 for a in self.attrs
553 if isinstance(a, (schema.Column, MappedColumn))
554 ]
556 @property
557 def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]:
558 return self
560 @property
561 def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
562 return [(c, 0) for c in self.columns if c.table is None]
564 @util.preload_module("orm.properties")
565 def _setup_arguments_on_columns(self) -> None:
566 """Propagate configuration arguments made on this composite
567 to the target columns, for those that apply.
569 """
570 ColumnProperty = util.preloaded.orm_properties.ColumnProperty
572 for prop in self.props:
573 if not isinstance(prop, ColumnProperty):
574 continue
575 else:
576 cprop = prop
578 cprop.active_history = self.active_history
579 if self.deferred:
580 cprop.deferred = self.deferred
581 cprop.strategy_key = (("deferred", True), ("instrument", True))
582 cprop.group = self.group
584 def _setup_event_handlers(self) -> None:
585 """Establish events that populate/expire the composite attribute."""
587 def load_handler(
588 state: InstanceState[Any], context: _ORMCompileState
589 ) -> None:
590 _load_refresh_handler(state, context, None, is_refresh=False)
592 def refresh_handler(
593 state: InstanceState[Any],
594 context: _ORMCompileState,
595 to_load: Optional[Sequence[str]],
596 ) -> None:
597 # note this corresponds to sqlalchemy.ext.mutable load_attrs()
599 if not to_load or (
600 {self.key}.union(self._attribute_keys)
601 ).intersection(to_load):
602 _load_refresh_handler(state, context, to_load, is_refresh=True)
604 def _load_refresh_handler(
605 state: InstanceState[Any],
606 context: _ORMCompileState,
607 to_load: Optional[Sequence[str]],
608 is_refresh: bool,
609 ) -> None:
610 dict_ = state.dict
612 # if context indicates we are coming from the
613 # fget() handler, this already set the value; skip the
614 # handler here. (other handlers like mutablecomposite will still
615 # want to catch it)
616 # there's an insufficiency here in that the fget() handler
617 # really should not be using the refresh event and there should
618 # be some other event that mutablecomposite can subscribe
619 # towards for this.
621 if (
622 not is_refresh or context is self._COMPOSITE_FGET
623 ) and self.key in dict_:
624 return
626 # if column elements aren't loaded, skip.
627 # __get__() will initiate a load for those
628 # columns
629 for k in self._attribute_keys:
630 if k not in dict_:
631 return
633 dict_[self.key] = self._construct_composite(
634 *[state.dict[key] for key in self._attribute_keys]
635 )
637 def expire_handler(
638 state: InstanceState[Any], keys: Optional[Sequence[str]]
639 ) -> None:
640 if keys is None or set(self._attribute_keys).intersection(keys):
641 state.dict.pop(self.key, None)
643 def insert_update_handler(
644 mapper: Mapper[Any],
645 connection: Connection,
646 state: InstanceState[Any],
647 ) -> None:
648 """After an insert or update, some columns may be expired due
649 to server side defaults, or re-populated due to client side
650 defaults. Pop out the composite value here so that it
651 recreates.
653 """
655 state.dict.pop(self.key, None)
657 event.listen(
658 self.parent, "after_insert", insert_update_handler, raw=True
659 )
660 event.listen(
661 self.parent, "after_update", insert_update_handler, raw=True
662 )
663 event.listen(
664 self.parent, "load", load_handler, raw=True, propagate=True
665 )
666 event.listen(
667 self.parent, "refresh", refresh_handler, raw=True, propagate=True
668 )
669 event.listen(
670 self.parent, "expire", expire_handler, raw=True, propagate=True
671 )
673 proxy_attr = self.parent.class_manager[self.key]
674 proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore
675 proxy_attr.impl.dispatch._active_history = self.active_history
677 # TODO: need a deserialize hook here
679 @util.memoized_property
680 def _attribute_keys(self) -> Sequence[str]:
681 return [prop.key for prop in self.props]
683 def _populate_composite_bulk_save_mappings_fn(
684 self,
685 ) -> Callable[[Dict[str, Any]], None]:
686 if self._generated_composite_accessor:
687 get_values = self._generated_composite_accessor
688 else:
690 def get_values(val: Any) -> Tuple[Any]:
691 return val.__composite_values__() # type: ignore
693 attrs = [prop.key for prop in self.props]
695 def populate(dest_dict: Dict[str, Any]) -> None:
696 dest_dict.update(
697 {
698 key: val
699 for key, val in zip(
700 attrs, get_values(dest_dict.pop(self.key))
701 )
702 }
703 )
705 return populate
707 def get_history(
708 self,
709 state: InstanceState[Any],
710 dict_: _InstanceDict,
711 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
712 ) -> History:
713 """Provided for userland code that uses attributes.get_history()."""
715 added: List[Any] = []
716 deleted: List[Any] = []
718 has_history = False
719 for prop in self.props:
720 key = prop.key
721 hist = state.manager[key].impl.get_history(state, dict_)
722 if hist.has_changes():
723 has_history = True
725 non_deleted = hist.non_deleted()
726 if non_deleted:
727 added.extend(non_deleted)
728 else:
729 added.append(None)
730 if hist.deleted:
731 deleted.extend(hist.deleted)
732 else:
733 deleted.append(None)
735 if has_history:
736 return attributes.History(
737 [self._construct_composite(*added)],
738 (),
739 [self._construct_composite(*deleted)],
740 )
741 else:
742 return attributes.History(
743 (), [self._construct_composite(*added)], ()
744 )
746 def _comparator_factory(
747 self, mapper: Mapper[Any]
748 ) -> Composite.Comparator[_CC]:
749 return self.comparator_factory(self, mapper)
751 class CompositeBundle(orm_util.Bundle[_T]):
752 def __init__(
753 self,
754 property_: Composite[_T],
755 expr: ClauseList,
756 ):
757 self.property = property_
758 super().__init__(property_.key, *expr)
760 def create_row_processor(
761 self,
762 query: Select[Unpack[TupleAny]],
763 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
764 labels: Sequence[str],
765 ) -> Callable[[Row[Unpack[TupleAny]]], Any]:
766 def proc(row: Row[Unpack[TupleAny]]) -> Any:
767 return self.property._construct_composite(
768 *[proc(row) for proc in procs]
769 )
771 return proc
773 class Comparator(PropComparator[_PT]):
774 """Produce boolean, comparison, and other operators for
775 :class:`.Composite` attributes.
777 See the example in :ref:`composite_operations` for an overview
778 of usage , as well as the documentation for :class:`.PropComparator`.
780 .. seealso::
782 :class:`.PropComparator`
784 :class:`.ColumnOperators`
786 :ref:`types_operators`
788 :attr:`.TypeEngine.comparator_factory`
790 """
792 # https://github.com/python/mypy/issues/4266
793 __hash__ = None # type: ignore
795 prop: RODescriptorReference[Composite[_PT]]
797 @util.memoized_property
798 def clauses(self) -> ClauseList:
799 return expression.ClauseList(
800 group=False, *self._comparable_elements
801 )
803 def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
804 return self.expression
806 @util.memoized_property
807 def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
808 clauses = self.clauses._annotate(
809 {
810 "parententity": self._parententity,
811 "parentmapper": self._parententity,
812 "proxy_key": self.prop.key,
813 }
814 )
815 return CompositeProperty.CompositeBundle(self.prop, clauses)
817 def _bulk_update_tuples(
818 self, value: Any
819 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
820 if isinstance(value, BindParameter):
821 value = value.value
823 values: Sequence[Any]
825 if value is None:
826 values = [None for key in self.prop._attribute_keys]
827 elif isinstance(self.prop.composite_class, type) and isinstance(
828 value, self.prop.composite_class
829 ):
830 values = self.prop._composite_values_from_instance(
831 value # type: ignore[arg-type]
832 )
833 else:
834 raise sa_exc.ArgumentError(
835 "Can't UPDATE composite attribute %s to %r"
836 % (self.prop, value)
837 )
839 return list(zip(self._comparable_elements, values))
841 def _bulk_dml_setter(self, key: str) -> Optional[Callable[..., Any]]:
842 return self.prop._populate_composite_bulk_save_mappings_fn()
844 @util.memoized_property
845 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
846 if self._adapt_to_entity:
847 return [
848 getattr(self._adapt_to_entity.entity, prop.key)
849 for prop in self.prop._comparable_elements
850 ]
851 else:
852 return self.prop._comparable_elements
854 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
855 return self._compare(operators.eq, other)
857 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
858 return self._compare(operators.ne, other)
860 def __lt__(self, other: Any) -> ColumnElement[bool]:
861 return self._compare(operators.lt, other)
863 def __gt__(self, other: Any) -> ColumnElement[bool]:
864 return self._compare(operators.gt, other)
866 def __le__(self, other: Any) -> ColumnElement[bool]:
867 return self._compare(operators.le, other)
869 def __ge__(self, other: Any) -> ColumnElement[bool]:
870 return self._compare(operators.ge, other)
872 def desc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
873 return expression.OrderByList(
874 [e.desc() for e in self._comparable_elements]
875 )
877 def asc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
878 return expression.OrderByList(
879 [e.asc() for e in self._comparable_elements]
880 )
882 def nulls_first(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
883 return expression.OrderByList(
884 [e.nulls_first() for e in self._comparable_elements]
885 )
887 def nulls_last(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
888 return expression.OrderByList(
889 [e.nulls_last() for e in self._comparable_elements]
890 )
892 # what might be interesting would be if we create
893 # an instance of the composite class itself with
894 # the columns as data members, then use "hybrid style" comparison
895 # to create these comparisons. then your Point.__eq__() method could
896 # be where comparison behavior is defined for SQL also. Likely
897 # not a good choice for default behavior though, not clear how it would
898 # work w/ dataclasses, etc. also no demand for any of this anyway.
899 def _compare(
900 self, operator: OperatorType, other: Any
901 ) -> ColumnElement[bool]:
902 values: Sequence[Any]
903 if other is None:
904 values = [None] * len(self.prop._comparable_elements)
905 else:
906 values = self.prop._composite_values_from_instance(other)
907 comparisons = [
908 operator(a, b)
909 for a, b in zip(self.prop._comparable_elements, values)
910 ]
911 if self._adapt_to_entity:
912 assert self.adapter is not None
913 comparisons = [self.adapter(x) for x in comparisons]
914 return sql.and_(*comparisons)
916 def __str__(self) -> str:
917 return str(self.parent.class_.__name__) + "." + self.key
920class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
921 """Declarative-compatible front-end for the :class:`.CompositeProperty`
922 class.
924 Public constructor is the :func:`_orm.composite` function.
926 .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
927 compatible subclass of :class:`_orm.CompositeProperty`.
929 .. seealso::
931 :ref:`mapper_composite`
933 """
935 inherit_cache = True
936 """:meta private:"""
939class ConcreteInheritedProperty(DescriptorProperty[_T]):
940 """A 'do nothing' :class:`.MapperProperty` that disables
941 an attribute on a concrete subclass that is only present
942 on the inherited mapper, not the concrete classes' mapper.
944 Cases where this occurs include:
946 * When the superclass mapper is mapped against a
947 "polymorphic union", which includes all attributes from
948 all subclasses.
949 * When a relationship() is configured on an inherited mapper,
950 but not on the subclass mapper. Concrete mappers require
951 that relationship() is configured explicitly on each
952 subclass.
954 """
956 def _comparator_factory(
957 self, mapper: Mapper[Any]
958 ) -> Type[PropComparator[_T]]:
959 comparator_callable = None
961 for m in self.parent.iterate_to_root():
962 p = m._props[self.key]
963 if getattr(p, "comparator_factory", None) is not None:
964 comparator_callable = p.comparator_factory
965 break
966 assert comparator_callable is not None
967 return comparator_callable(p, mapper) # type: ignore
969 def __init__(self) -> None:
970 super().__init__()
972 def warn() -> NoReturn:
973 raise AttributeError(
974 "Concrete %s does not implement "
975 "attribute %r at the instance level. Add "
976 "this property explicitly to %s."
977 % (self.parent, self.key, self.parent)
978 )
980 class NoninheritedConcreteProp:
981 def __set__(s: Any, obj: Any, value: Any) -> NoReturn:
982 warn()
984 def __delete__(s: Any, obj: Any) -> NoReturn:
985 warn()
987 def __get__(s: Any, obj: Any, owner: Any) -> Any:
988 if obj is None:
989 return self.descriptor
990 warn()
992 self.descriptor = NoninheritedConcreteProp()
995class SynonymProperty(DescriptorProperty[_T]):
996 """Denote an attribute name as a synonym to a mapped property,
997 in that the attribute will mirror the value and expression behavior
998 of another attribute.
1000 :class:`.Synonym` is constructed using the :func:`_orm.synonym`
1001 function.
1003 .. seealso::
1005 :ref:`synonyms` - Overview of synonyms
1007 """
1009 comparator_factory: Optional[Type[PropComparator[_T]]]
1011 def __init__(
1012 self,
1013 name: str,
1014 map_column: Optional[bool] = None,
1015 descriptor: Optional[Any] = None,
1016 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
1017 attribute_options: Optional[_AttributeOptions] = None,
1018 info: Optional[_InfoType] = None,
1019 doc: Optional[str] = None,
1020 ):
1021 super().__init__(attribute_options=attribute_options)
1023 self.name = name
1024 self.map_column = map_column
1025 self.descriptor = descriptor
1026 self.comparator_factory = comparator_factory
1027 if doc:
1028 self.doc = doc
1029 elif descriptor and descriptor.__doc__:
1030 self.doc = descriptor.__doc__
1031 else:
1032 self.doc = None
1033 if info:
1034 self.info.update(info)
1036 util.set_creation_order(self)
1038 if not TYPE_CHECKING:
1040 @property
1041 def uses_objects(self) -> bool:
1042 return getattr(self.parent.class_, self.name).impl.uses_objects
1044 # TODO: when initialized, check _proxied_object,
1045 # emit a warning if its not a column-based property
1047 @util.memoized_property
1048 def _proxied_object(
1049 self,
1050 ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]:
1051 attr = getattr(self.parent.class_, self.name)
1052 if not hasattr(attr, "property") or not isinstance(
1053 attr.property, MapperProperty
1054 ):
1055 # attribute is a non-MapperProprerty proxy such as
1056 # hybrid or association proxy
1057 if isinstance(attr, attributes.QueryableAttribute):
1058 return attr.comparator
1059 elif isinstance(attr, SQLORMOperations):
1060 # assocaition proxy comes here
1061 return attr
1063 raise sa_exc.InvalidRequestError(
1064 """synonym() attribute "%s.%s" only supports """
1065 """ORM mapped attributes, got %r"""
1066 % (self.parent.class_.__name__, self.name, attr)
1067 )
1068 return attr.property
1070 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
1071 return (getattr(self.parent.class_, self.name),)
1073 def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]:
1074 prop = self._proxied_object
1076 if isinstance(prop, MapperProperty):
1077 if self.comparator_factory:
1078 comp = self.comparator_factory(prop, mapper)
1079 else:
1080 comp = prop.comparator_factory(prop, mapper)
1081 return comp
1082 else:
1083 return prop
1085 def get_history(
1086 self,
1087 state: InstanceState[Any],
1088 dict_: _InstanceDict,
1089 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1090 ) -> History:
1091 attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name)
1092 return attr.impl.get_history(state, dict_, passive=passive)
1094 def _get_dataclass_setup_options(
1095 self,
1096 decl_scan: _ClassScanAbstractConfig,
1097 key: str,
1098 dataclass_setup_arguments: _DataclassArguments,
1099 enable_descriptor_defaults: bool,
1100 ) -> _AttributeOptions:
1101 dataclasses_default = self._attribute_options.dataclasses_default
1102 if (
1103 dataclasses_default is not _NoArg.NO_ARG
1104 and not callable(dataclasses_default)
1105 and enable_descriptor_defaults
1106 and not getattr(
1107 decl_scan.cls, "_sa_disable_descriptor_defaults", False
1108 )
1109 ):
1110 proxied = decl_scan.collected_attributes[self.name]
1111 proxied_default = proxied._attribute_options.dataclasses_default
1112 if proxied_default != dataclasses_default:
1113 raise sa_exc.ArgumentError(
1114 f"Synonym {key!r} default argument "
1115 f"{dataclasses_default!r} must match the dataclasses "
1116 f"default value of proxied object {self.name!r}, "
1117 f"""currently {
1118 repr(proxied_default)
1119 if proxied_default is not _NoArg.NO_ARG
1120 else 'not set'}"""
1121 )
1122 self._default_scalar_value = dataclasses_default
1123 return self._attribute_options._replace(
1124 dataclasses_default=DONT_SET
1125 )
1127 return self._attribute_options
1129 @util.preload_module("sqlalchemy.orm.properties")
1130 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
1131 properties = util.preloaded.orm_properties
1133 if self.map_column:
1134 # implement the 'map_column' option.
1135 if self.key not in parent.persist_selectable.c:
1136 raise sa_exc.ArgumentError(
1137 "Can't compile synonym '%s': no column on table "
1138 "'%s' named '%s'"
1139 % (
1140 self.name,
1141 parent.persist_selectable.description,
1142 self.key,
1143 )
1144 )
1145 elif (
1146 parent.persist_selectable.c[self.key]
1147 in parent._columntoproperty
1148 and parent._columntoproperty[
1149 parent.persist_selectable.c[self.key]
1150 ].key
1151 == self.name
1152 ):
1153 raise sa_exc.ArgumentError(
1154 "Can't call map_column=True for synonym %r=%r, "
1155 "a ColumnProperty already exists keyed to the name "
1156 "%r for column %r"
1157 % (self.key, self.name, self.name, self.key)
1158 )
1159 p: ColumnProperty[Any] = properties.ColumnProperty(
1160 parent.persist_selectable.c[self.key]
1161 )
1162 parent._configure_property(self.name, p, init=init, setparent=True)
1163 p._mapped_by_synonym = self.key
1165 self.parent = parent
1168class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
1169 """Declarative front-end for the :class:`.SynonymProperty` class.
1171 Public constructor is the :func:`_orm.synonym` function.
1173 .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
1174 compatible subclass for :class:`_orm.SynonymProperty`
1176 .. seealso::
1178 :ref:`synonyms` - Overview of synonyms
1180 """
1182 inherit_cache = True
1183 """:meta private:"""