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-2026 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"""
14from __future__ import annotations
16from dataclasses import is_dataclass
17import inspect
18import itertools
19import operator
20import typing
21from typing import Any
22from typing import Callable
23from typing import Dict
24from typing import get_args
25from typing import List
26from typing import NoReturn
27from typing import Optional
28from typing import Sequence
29from typing import Tuple
30from typing import Type
31from typing import TYPE_CHECKING
32from typing import TypeVar
33from typing import Union
34import weakref
36from . import attributes
37from . import util as orm_util
38from .base import _DeclarativeMapped
39from .base import DONT_SET
40from .base import LoaderCallableStatus
41from .base import Mapped
42from .base import PassiveFlag
43from .base import SQLORMOperations
44from .interfaces import _AttributeOptions
45from .interfaces import _IntrospectsAnnotations
46from .interfaces import _MapsColumns
47from .interfaces import MapperProperty
48from .interfaces import PropComparator
49from .util import de_stringify_annotation
50from .. import event
51from .. import exc as sa_exc
52from .. import schema
53from .. import sql
54from .. import util
55from ..sql import expression
56from ..sql import operators
57from ..sql.base import _NoArg
58from ..sql.elements import BindParameter
59from ..util.typing import de_optionalize_union_types
60from ..util.typing import includes_none
61from ..util.typing import is_fwd_ref
62from ..util.typing import is_pep593
63from ..util.typing import is_union
64from ..util.typing import TupleAny
65from ..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[override]
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], ...]
211 column_template: Optional[str]
213 _generated_composite_accessor: CallableReference[
214 Optional[Callable[[_CC], Tuple[Any, ...]]]
215 ]
217 comparator_factory: Type[Comparator[_CC]]
219 def __init__(
220 self,
221 _class_or_attr: Union[
222 None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any]
223 ] = None,
224 *attrs: _CompositeAttrType[Any],
225 return_none_on: Union[
226 _NoArg, None, Callable[..., bool]
227 ] = _NoArg.NO_ARG,
228 attribute_options: Optional[_AttributeOptions] = None,
229 active_history: bool = False,
230 deferred: bool = False,
231 group: Optional[str] = None,
232 comparator_factory: Optional[Type[Comparator[_CC]]] = None,
233 column_template: Optional[str] = None,
234 info: Optional[_InfoType] = None,
235 **kwargs: Any,
236 ):
237 super().__init__(attribute_options=attribute_options)
239 if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
240 self.attrs = (_class_or_attr,) + attrs
241 # will initialize within declarative_scan
242 self.composite_class = None # type: ignore[assignment]
243 else:
244 self.composite_class = _class_or_attr # type: ignore[assignment]
245 self.attrs = attrs
247 if column_template is not None:
248 try:
249 column_template % "x"
250 except (TypeError, ValueError) as te:
251 raise sa_exc.ArgumentError(
252 f"column_template {column_template!r} is not a valid "
253 "template; expected a string containing exactly one "
254 "'%s' placeholder"
255 ) from te
256 self.column_template = column_template
258 self.return_none_on = return_none_on
259 self.active_history = active_history
260 self.deferred = deferred
261 self.group = group
262 self.comparator_factory = (
263 comparator_factory
264 if comparator_factory is not None
265 else self.__class__.Comparator
266 )
267 self._generated_composite_accessor = None
268 if info is not None:
269 self.info.update(info)
271 util.set_creation_order(self)
272 self._create_descriptor()
273 self._init_accessor()
275 @util.memoized_property
276 def _construct_composite(self) -> Callable[..., Any]:
277 return_none_on = self.return_none_on
278 if callable(return_none_on):
280 def construct(*args: Any) -> Any:
281 if return_none_on(*args):
282 return None
283 else:
284 return self.composite_class(*args)
286 return construct
287 else:
288 return self.composite_class
290 def instrument_class(self, mapper: Mapper[Any]) -> None:
291 super().instrument_class(mapper)
292 self._setup_event_handlers()
294 def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]:
295 if self._generated_composite_accessor:
296 return self._generated_composite_accessor(value)
297 else:
298 try:
299 accessor = value.__composite_values__
300 except AttributeError as ae:
301 raise sa_exc.InvalidRequestError(
302 f"Composite class {self.composite_class.__name__} is not "
303 f"a dataclass and does not define a __composite_values__()"
304 " method; can't get state"
305 ) from ae
306 else:
307 return accessor() # type: ignore[no-any-return]
309 def do_init(self) -> None:
310 """Initialization which occurs after the :class:`.Composite`
311 has been associated with its parent mapper.
313 """
314 self._setup_arguments_on_columns()
316 _COMPOSITE_FGET = object()
318 def _create_descriptor(self) -> None:
319 """Create the Python descriptor that will serve as
320 the access point on instances of the mapped class.
322 """
324 def fget(instance: Any) -> Any:
325 dict_ = attributes.instance_dict(instance)
326 state = attributes.instance_state(instance)
328 if self.key not in dict_:
329 # key not present. Iterate through related
330 # attributes, retrieve their values. This
331 # ensures they all load.
332 values = [
333 getattr(instance, key) for key in self._attribute_keys
334 ]
336 if self.key not in dict_:
337 dict_[self.key] = self._construct_composite(*values)
338 state.manager.dispatch.refresh(
339 state, self._COMPOSITE_FGET, [self.key]
340 )
342 return dict_.get(self.key, None)
344 def fset(instance: Any, value: Any) -> None:
345 if value is LoaderCallableStatus.DONT_SET:
346 return
348 dict_ = attributes.instance_dict(instance)
349 state = attributes.instance_state(instance)
350 attr = state.manager[self.key]
352 if attr.dispatch._active_history:
353 previous = fget(instance)
354 else:
355 previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
357 for fn in attr.dispatch.set:
358 value = fn(state, value, previous, attr.impl)
359 dict_[self.key] = value
360 if value is None:
361 for key in self._attribute_keys:
362 setattr(instance, key, None)
363 else:
364 for key, value in zip(
365 self._attribute_keys,
366 self._composite_values_from_instance(value),
367 ):
368 setattr(instance, key, value)
370 def fdel(instance: Any) -> None:
371 state = attributes.instance_state(instance)
372 dict_ = attributes.instance_dict(instance)
373 attr = state.manager[self.key]
375 if attr.dispatch._active_history:
376 previous = fget(instance)
377 dict_.pop(self.key, None)
378 else:
379 previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
381 attr = state.manager[self.key]
382 attr.dispatch.remove(state, previous, attr.impl)
383 for key in self._attribute_keys:
384 setattr(instance, key, None)
386 self.descriptor = property(fget, fset, fdel)
388 @util.preload_module("sqlalchemy.orm.properties")
389 def declarative_scan(
390 self,
391 decl_scan: _DeclarativeMapperConfig,
392 registry: _RegistryType,
393 cls: Type[Any],
394 originating_module: Optional[str],
395 key: str,
396 mapped_container: Optional[Type[Mapped[Any]]],
397 annotation: Optional[_AnnotationScanType],
398 extracted_mapped_annotation: Optional[_AnnotationScanType],
399 is_dataclass_field: bool,
400 ) -> None:
401 MappedColumn = util.preloaded.orm_properties.MappedColumn
402 if (
403 self.composite_class is None
404 and extracted_mapped_annotation is None
405 ):
406 self._raise_for_required(key, cls)
407 argument = extracted_mapped_annotation
409 if is_pep593(argument):
410 argument = get_args(argument)[0]
412 if argument and self.composite_class is None:
413 if isinstance(argument, str) or is_fwd_ref(
414 argument, check_generic=True
415 ):
416 if originating_module is None:
417 str_arg = (
418 argument.__forward_arg__
419 if hasattr(argument, "__forward_arg__")
420 else str(argument)
421 )
422 raise sa_exc.ArgumentError(
423 f"Can't use forward ref {argument} for composite "
424 f"class argument; set up the type as Mapped[{str_arg}]"
425 )
426 argument = de_stringify_annotation(
427 cls, argument, originating_module, include_generic=True
428 )
430 if is_union(argument) and includes_none(argument):
431 if self.return_none_on is _NoArg.NO_ARG:
432 self.return_none_on = lambda *args: all(
433 arg is None for arg in args
434 )
435 argument = de_optionalize_union_types(argument)
437 self.composite_class = argument
439 if is_dataclass(self.composite_class):
440 self._setup_for_dataclass(
441 decl_scan, registry, cls, originating_module, key
442 )
443 else:
444 if self.column_template is not None:
445 raise sa_exc.ArgumentError(
446 "column_template is only supported when composite_class "
447 "is a dataclass"
448 )
449 for attr in self.attrs:
450 if (
451 isinstance(attr, (MappedColumn, schema.Column))
452 and attr.name is None
453 ):
454 raise sa_exc.ArgumentError(
455 "Composite class column arguments must be named "
456 "unless a dataclass is used"
457 )
458 self._init_accessor()
460 def _init_accessor(self) -> None:
461 if is_dataclass(self.composite_class) and not hasattr(
462 self.composite_class, "__composite_values__"
463 ):
464 insp = inspect.signature(self.composite_class)
465 getter = operator.attrgetter(
466 *[p.name for p in insp.parameters.values()]
467 )
468 if len(insp.parameters) == 1:
469 self._generated_composite_accessor = lambda obj: (getter(obj),)
470 else:
471 self._generated_composite_accessor = getter
473 if (
474 self.composite_class is not None
475 and isinstance(self.composite_class, type)
476 and self.composite_class not in _composite_getters
477 ):
478 if self._generated_composite_accessor is not None:
479 _composite_getters[self.composite_class] = (
480 self._generated_composite_accessor
481 )
482 elif hasattr(self.composite_class, "__composite_values__"):
483 _composite_getters[self.composite_class] = (
484 lambda obj: obj.__composite_values__()
485 )
487 @util.preload_module("sqlalchemy.orm.properties")
488 @util.preload_module("sqlalchemy.orm.decl_base")
489 def _setup_for_dataclass(
490 self,
491 decl_scan: _DeclarativeMapperConfig,
492 registry: _RegistryType,
493 cls: Type[Any],
494 originating_module: Optional[str],
495 key: str,
496 ) -> None:
497 MappedColumn = util.preloaded.orm_properties.MappedColumn
499 decl_base = util.preloaded.orm_decl_base
501 insp = inspect.signature(self.composite_class)
502 for param, attr in itertools.zip_longest(
503 insp.parameters.values(), self.attrs
504 ):
505 if param is None:
506 raise sa_exc.ArgumentError(
507 f"number of composite attributes "
508 f"{len(self.attrs)} exceeds "
509 f"that of the number of attributes in class "
510 f"{self.composite_class.__name__} {len(insp.parameters)}"
511 )
512 if attr is None:
513 # fill in missing attr spots with empty MappedColumn,
514 # or one named from column_template if present
515 if self.column_template is not None:
516 attr = MappedColumn(self.column_template % param.name)
517 else:
518 attr = MappedColumn()
519 self.attrs += (attr,)
521 if isinstance(attr, MappedColumn):
522 attr.declarative_scan_for_composite(
523 decl_scan,
524 registry,
525 cls,
526 originating_module,
527 key,
528 param.name,
529 param.annotation,
530 )
531 elif isinstance(attr, schema.Column):
532 decl_base._undefer_column_name(param.name, attr)
534 @util.memoized_property
535 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
536 return [getattr(self.parent.class_, prop.key) for prop in self.props]
538 @util.memoized_property
539 @util.preload_module("orm.properties")
540 def props(self) -> Sequence[MapperProperty[Any]]:
541 props = []
542 MappedColumn = util.preloaded.orm_properties.MappedColumn
544 for attr in self.attrs:
545 if isinstance(attr, str):
546 prop = self.parent.get_property(attr, _configure_mappers=False)
547 elif isinstance(attr, schema.Column):
548 prop = self.parent._columntoproperty[attr]
549 elif isinstance(attr, MappedColumn):
550 prop = self.parent._columntoproperty[attr.column]
551 elif isinstance(attr, attributes.InstrumentedAttribute):
552 prop = attr.property
553 else:
554 prop = None
556 if not isinstance(prop, MapperProperty):
557 raise sa_exc.ArgumentError(
558 "Composite expects Column objects or mapped "
559 f"attributes/attribute names as arguments, got: {attr!r}"
560 )
562 props.append(prop)
563 return props
565 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
566 return self._comparable_elements
568 @util.non_memoized_property
569 @util.preload_module("orm.properties")
570 def columns(self) -> Sequence[Column[Any]]:
571 MappedColumn = util.preloaded.orm_properties.MappedColumn
572 return [
573 a.column if isinstance(a, MappedColumn) else a
574 for a in self.attrs
575 if isinstance(a, (schema.Column, MappedColumn))
576 ]
578 @property
579 def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]:
580 return self
582 @property
583 def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
584 return [(c, 0) for c in self.columns if c.table is None]
586 @util.preload_module("orm.properties")
587 def _setup_arguments_on_columns(self) -> None:
588 """Propagate configuration arguments made on this composite
589 to the target columns, for those that apply.
591 """
592 ColumnProperty = util.preloaded.orm_properties.ColumnProperty
594 for prop in self.props:
595 if not isinstance(prop, ColumnProperty):
596 continue
597 else:
598 cprop = prop
600 cprop.active_history = self.active_history
601 if self.deferred:
602 cprop.deferred = self.deferred
603 cprop.strategy_key = (("deferred", True), ("instrument", True))
604 cprop.group = self.group
606 def _setup_event_handlers(self) -> None:
607 """Establish events that populate/expire the composite attribute."""
609 def load_handler(
610 state: InstanceState[Any], context: _ORMCompileState
611 ) -> None:
612 _load_refresh_handler(state, context, None, is_refresh=False)
614 def refresh_handler(
615 state: InstanceState[Any],
616 context: _ORMCompileState,
617 to_load: Optional[Sequence[str]],
618 ) -> None:
619 # note this corresponds to sqlalchemy.ext.mutable load_attrs()
621 if not to_load or (
622 {self.key}.union(self._attribute_keys)
623 ).intersection(to_load):
624 _load_refresh_handler(state, context, to_load, is_refresh=True)
626 def _load_refresh_handler(
627 state: InstanceState[Any],
628 context: _ORMCompileState,
629 to_load: Optional[Sequence[str]],
630 is_refresh: bool,
631 ) -> None:
632 dict_ = state.dict
634 # if context indicates we are coming from the
635 # fget() handler, this already set the value; skip the
636 # handler here. (other handlers like mutablecomposite will still
637 # want to catch it)
638 # there's an insufficiency here in that the fget() handler
639 # really should not be using the refresh event and there should
640 # be some other event that mutablecomposite can subscribe
641 # towards for this.
643 if (
644 not is_refresh or context is self._COMPOSITE_FGET
645 ) and self.key in dict_:
646 return
648 # if column elements aren't loaded, skip.
649 # __get__() will initiate a load for those
650 # columns
651 for k in self._attribute_keys:
652 if k not in dict_:
653 return
655 dict_[self.key] = self._construct_composite(
656 *[state.dict[key] for key in self._attribute_keys]
657 )
659 def expire_handler(
660 state: InstanceState[Any], keys: Optional[Sequence[str]]
661 ) -> None:
662 if keys is None or set(self._attribute_keys).intersection(keys):
663 state.dict.pop(self.key, None)
665 def insert_update_handler(
666 mapper: Mapper[Any],
667 connection: Connection,
668 state: InstanceState[Any],
669 ) -> None:
670 """After an insert or update, some columns may be expired due
671 to server side defaults, or re-populated due to client side
672 defaults. Pop out the composite value here so that it
673 recreates.
675 """
677 state.dict.pop(self.key, None)
679 event.listen(
680 self.parent, "after_insert", insert_update_handler, raw=True
681 )
682 event.listen(
683 self.parent, "after_update", insert_update_handler, raw=True
684 )
685 event.listen(
686 self.parent, "load", load_handler, raw=True, propagate=True
687 )
688 event.listen(
689 self.parent, "refresh", refresh_handler, raw=True, propagate=True
690 )
691 event.listen(
692 self.parent, "expire", expire_handler, raw=True, propagate=True
693 )
695 proxy_attr = self.parent.class_manager[self.key]
696 proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore[assignment] # noqa: E501
697 proxy_attr.impl.dispatch._active_history = self.active_history
699 # TODO: need a deserialize hook here
701 @util.memoized_property
702 def _attribute_keys(self) -> Sequence[str]:
703 return [prop.key for prop in self.props]
705 def _populate_composite_bulk_save_mappings_fn(
706 self,
707 ) -> Callable[[Dict[str, Any]], None]:
708 if self._generated_composite_accessor:
709 get_values = self._generated_composite_accessor
710 else:
712 def get_values(val: Any) -> Tuple[Any]:
713 return val.__composite_values__() # type: ignore[no-any-return] # noqa: E501
715 attrs = [prop.key for prop in self.props]
717 def populate(dest_dict: Dict[str, Any]) -> None:
718 dest_dict.update(
719 {
720 key: val
721 for key, val in zip(
722 attrs, get_values(dest_dict.pop(self.key))
723 )
724 }
725 )
727 return populate
729 def get_history(
730 self,
731 state: InstanceState[Any],
732 dict_: _InstanceDict,
733 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
734 ) -> History:
735 """Provided for userland code that uses attributes.get_history()."""
737 added: List[Any] = []
738 deleted: List[Any] = []
740 has_history = False
741 for prop in self.props:
742 key = prop.key
743 hist = state.manager[key].impl.get_history(state, dict_)
744 if hist.has_changes():
745 has_history = True
747 non_deleted = hist.non_deleted()
748 if non_deleted:
749 added.extend(non_deleted)
750 else:
751 added.append(None)
752 if hist.deleted:
753 deleted.extend(hist.deleted)
754 else:
755 deleted.append(None)
757 if has_history:
758 return attributes.History(
759 [self._construct_composite(*added)],
760 (),
761 [self._construct_composite(*deleted)],
762 )
763 else:
764 return attributes.History(
765 (), [self._construct_composite(*added)], ()
766 )
768 def _comparator_factory(
769 self, mapper: Mapper[Any]
770 ) -> Composite.Comparator[_CC]:
771 return self.comparator_factory(self, mapper)
773 class CompositeBundle(orm_util.Bundle[_T]):
774 def __init__(
775 self,
776 property_: Composite[_T],
777 expr: ClauseList,
778 ):
779 self.property = property_
780 super().__init__(property_.key, *expr)
782 def create_row_processor(
783 self,
784 query: Select[Unpack[TupleAny]],
785 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
786 labels: Sequence[str],
787 ) -> Callable[[Row[Unpack[TupleAny]]], Any]:
788 def proc(row: Row[Unpack[TupleAny]]) -> Any:
789 return self.property._construct_composite(
790 *[proc(row) for proc in procs]
791 )
793 return proc
795 class Comparator(PropComparator[_PT]):
796 """Produce boolean, comparison, and other operators for
797 :class:`.Composite` attributes.
799 See the example in :ref:`composite_operations` for an overview
800 of usage , as well as the documentation for :class:`.PropComparator`.
802 .. seealso::
804 :class:`.PropComparator`
806 :class:`.ColumnOperators`
808 :ref:`types_operators`
810 :attr:`.TypeEngine.comparator_factory`
812 """
814 # https://github.com/python/mypy/issues/4266
815 __hash__ = None # type: ignore[assignment]
817 prop: RODescriptorReference[Composite[_PT]]
819 @util.memoized_property
820 def clauses(self) -> ClauseList:
821 return expression.ClauseList(
822 group=False, *self._comparable_elements
823 )
825 def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
826 return self.expression
828 @util.memoized_property
829 def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
830 clauses = self.clauses._annotate(
831 {
832 "parententity": self._parententity,
833 "parentmapper": self._parententity,
834 "proxy_key": self.prop.key,
835 }
836 )
837 return CompositeProperty.CompositeBundle(self.prop, clauses)
839 def _bulk_update_tuples(
840 self, value: Any
841 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
842 if isinstance(value, BindParameter):
843 value = value.value
845 values: Sequence[Any]
847 if value is None:
848 values = [None for key in self.prop._attribute_keys]
849 elif isinstance(self.prop.composite_class, type) and isinstance(
850 value, self.prop.composite_class
851 ):
852 values = self.prop._composite_values_from_instance(
853 value # type: ignore[arg-type]
854 )
855 else:
856 raise sa_exc.ArgumentError(
857 "Can't UPDATE composite attribute %s to %r"
858 % (self.prop, value)
859 )
861 return list(zip(self._comparable_elements, values))
863 def _bulk_dml_setter(self, key: str) -> Optional[Callable[..., Any]]:
864 return self.prop._populate_composite_bulk_save_mappings_fn()
866 @util.memoized_property
867 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
868 if self._adapt_to_entity:
869 return [
870 getattr(self._adapt_to_entity.entity, prop.key)
871 for prop in self.prop._comparable_elements
872 ]
873 else:
874 return self.prop._comparable_elements
876 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
877 return self._compare(operators.eq, other)
879 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
880 return self._compare(operators.ne, other)
882 def __lt__(self, other: Any) -> ColumnElement[bool]:
883 return self._compare(operators.lt, other)
885 def __gt__(self, other: Any) -> ColumnElement[bool]:
886 return self._compare(operators.gt, other)
888 def __le__(self, other: Any) -> ColumnElement[bool]:
889 return self._compare(operators.le, other)
891 def __ge__(self, other: Any) -> ColumnElement[bool]:
892 return self._compare(operators.ge, other)
894 def desc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
895 return expression.OrderByList(
896 [e.desc() for e in self._comparable_elements]
897 )
899 def asc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
900 return expression.OrderByList(
901 [e.asc() for e in self._comparable_elements]
902 )
904 def nulls_first(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
905 return expression.OrderByList(
906 [e.nulls_first() for e in self._comparable_elements]
907 )
909 def nulls_last(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
910 return expression.OrderByList(
911 [e.nulls_last() for e in self._comparable_elements]
912 )
914 # what might be interesting would be if we create
915 # an instance of the composite class itself with
916 # the columns as data members, then use "hybrid style" comparison
917 # to create these comparisons. then your Point.__eq__() method could
918 # be where comparison behavior is defined for SQL also. Likely
919 # not a good choice for default behavior though, not clear how it would
920 # work w/ dataclasses, etc. also no demand for any of this anyway.
921 def _compare(
922 self, operator: OperatorType, other: Any
923 ) -> ColumnElement[bool]:
924 values: Sequence[Any]
925 if other is None:
926 values = [None] * len(self.prop._comparable_elements)
927 else:
928 values = self.prop._composite_values_from_instance(other)
929 comparisons = [
930 operator(a, b)
931 for a, b in zip(self.prop._comparable_elements, values)
932 ]
933 if self._adapt_to_entity:
934 assert self.adapter is not None
935 comparisons = [self.adapter(x) for x in comparisons]
936 return sql.and_(*comparisons)
938 def __str__(self) -> str:
939 return str(self.parent.class_.__name__) + "." + self.key
942class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
943 """Declarative-compatible front-end for the :class:`.CompositeProperty`
944 class.
946 Public constructor is the :func:`_orm.composite` function.
948 .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
949 compatible subclass of :class:`_orm.CompositeProperty`.
951 .. seealso::
953 :ref:`mapper_composite`
955 """
957 inherit_cache = True
958 """:meta private:"""
961class ConcreteInheritedProperty(DescriptorProperty[_T]):
962 """A 'do nothing' :class:`.MapperProperty` that disables
963 an attribute on a concrete subclass that is only present
964 on the inherited mapper, not the concrete classes' mapper.
966 Cases where this occurs include:
968 * When the superclass mapper is mapped against a
969 "polymorphic union", which includes all attributes from
970 all subclasses.
971 * When a relationship() is configured on an inherited mapper,
972 but not on the subclass mapper. Concrete mappers require
973 that relationship() is configured explicitly on each
974 subclass.
976 """
978 def _comparator_factory(
979 self, mapper: Mapper[Any]
980 ) -> Type[PropComparator[_T]]:
981 comparator_callable = None
983 for m in self.parent.iterate_to_root():
984 p = m._props[self.key]
985 if getattr(p, "comparator_factory", None) is not None:
986 comparator_callable = p.comparator_factory
987 break
988 assert comparator_callable is not None
989 return comparator_callable(p, mapper) # type: ignore[no-any-return]
991 def __init__(self) -> None:
992 super().__init__()
994 def warn() -> NoReturn:
995 raise AttributeError(
996 "Concrete %s does not implement "
997 "attribute %r at the instance level. Add "
998 "this property explicitly to %s."
999 % (self.parent, self.key, self.parent)
1000 )
1002 class NoninheritedConcreteProp:
1003 def __set__(s: Any, obj: Any, value: Any) -> NoReturn:
1004 warn()
1006 def __delete__(s: Any, obj: Any) -> NoReturn:
1007 warn()
1009 def __get__(s: Any, obj: Any, owner: Any) -> Any:
1010 if obj is None:
1011 return self.descriptor
1012 warn()
1014 self.descriptor = NoninheritedConcreteProp()
1017class SynonymProperty(DescriptorProperty[_T]):
1018 """Denote an attribute name as a synonym to a mapped property,
1019 in that the attribute will mirror the value and expression behavior
1020 of another attribute.
1022 :class:`.Synonym` is constructed using the :func:`_orm.synonym`
1023 function.
1025 .. seealso::
1027 :ref:`synonyms` - Overview of synonyms
1029 """
1031 comparator_factory: Optional[Type[PropComparator[_T]]]
1033 def __init__(
1034 self,
1035 name: str,
1036 map_column: Optional[bool] = None,
1037 descriptor: Optional[Any] = None,
1038 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
1039 attribute_options: Optional[_AttributeOptions] = None,
1040 info: Optional[_InfoType] = None,
1041 doc: Optional[str] = None,
1042 ):
1043 super().__init__(attribute_options=attribute_options)
1045 self.name = name
1046 self.map_column = map_column
1047 self.descriptor = descriptor
1048 self.comparator_factory = comparator_factory
1049 if doc:
1050 self.doc = doc
1051 elif descriptor and descriptor.__doc__:
1052 self.doc = descriptor.__doc__
1053 else:
1054 self.doc = None
1055 if info:
1056 self.info.update(info)
1058 util.set_creation_order(self)
1060 if not TYPE_CHECKING:
1062 @property
1063 def uses_objects(self) -> bool:
1064 return getattr(self.parent.class_, self.name).impl.uses_objects
1066 # TODO: when initialized, check _proxied_object,
1067 # emit a warning if its not a column-based property
1069 @util.memoized_property
1070 def _proxied_object(
1071 self,
1072 ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]:
1073 attr = getattr(self.parent.class_, self.name)
1074 if not hasattr(attr, "property") or not isinstance(
1075 attr.property, MapperProperty
1076 ):
1077 # attribute is a non-MapperProprerty proxy such as
1078 # hybrid or association proxy
1079 if isinstance(attr, attributes.QueryableAttribute):
1080 return attr.comparator
1081 elif isinstance(attr, SQLORMOperations):
1082 # association proxy comes here
1083 return attr
1085 raise sa_exc.InvalidRequestError(
1086 """synonym() attribute "%s.%s" only supports """
1087 """ORM mapped attributes, got %r"""
1088 % (self.parent.class_.__name__, self.name, attr)
1089 )
1090 return attr.property
1092 def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]:
1093 return (getattr(self.parent.class_, self.name),)
1095 def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]:
1096 prop = self._proxied_object
1098 if isinstance(prop, MapperProperty):
1099 if self.comparator_factory:
1100 comp = self.comparator_factory(prop, mapper)
1101 else:
1102 comp = prop.comparator_factory(prop, mapper)
1103 return comp
1104 else:
1105 return prop
1107 def get_history(
1108 self,
1109 state: InstanceState[Any],
1110 dict_: _InstanceDict,
1111 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1112 ) -> History:
1113 attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name)
1114 return attr.impl.get_history(state, dict_, passive=passive)
1116 def _get_dataclass_setup_options(
1117 self,
1118 decl_scan: _ClassScanAbstractConfig,
1119 key: str,
1120 dataclass_setup_arguments: _DataclassArguments,
1121 enable_descriptor_defaults: bool,
1122 ) -> _AttributeOptions:
1123 dataclasses_default = self._attribute_options.dataclasses_default
1124 if (
1125 dataclasses_default is not _NoArg.NO_ARG
1126 and not callable(dataclasses_default)
1127 and enable_descriptor_defaults
1128 and not getattr(
1129 decl_scan.cls, "_sa_disable_descriptor_defaults", False
1130 )
1131 ):
1132 proxied = decl_scan.collected_attributes[self.name]
1133 proxied_default = proxied._attribute_options.dataclasses_default
1134 if proxied_default != dataclasses_default:
1135 raise sa_exc.ArgumentError(
1136 f"Synonym {key!r} default argument "
1137 f"{dataclasses_default!r} must match the dataclasses "
1138 f"default value of proxied object {self.name!r}, "
1139 f"""currently {
1140 repr(proxied_default)
1141 if proxied_default is not _NoArg.NO_ARG
1142 else 'not set'}"""
1143 )
1144 self._default_scalar_value = dataclasses_default
1145 return self._attribute_options._replace(
1146 dataclasses_default=DONT_SET
1147 )
1149 return self._attribute_options
1151 @util.preload_module("sqlalchemy.orm.properties")
1152 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
1153 properties = util.preloaded.orm_properties
1155 if self.map_column:
1156 # implement the 'map_column' option.
1157 if self.key not in parent.persist_selectable.c:
1158 raise sa_exc.ArgumentError(
1159 "Can't compile synonym '%s': no column on table "
1160 "'%s' named '%s'"
1161 % (
1162 self.name,
1163 parent.persist_selectable.description,
1164 self.key,
1165 )
1166 )
1167 elif (
1168 parent.persist_selectable.c[self.key]
1169 in parent._columntoproperty
1170 and parent._columntoproperty[
1171 parent.persist_selectable.c[self.key]
1172 ].key
1173 == self.name
1174 ):
1175 raise sa_exc.ArgumentError(
1176 "Can't call map_column=True for synonym %r=%r, "
1177 "a ColumnProperty already exists keyed to the name "
1178 "%r for column %r"
1179 % (self.key, self.name, self.name, self.key)
1180 )
1181 p: ColumnProperty[Any] = properties.ColumnProperty(
1182 parent.persist_selectable.c[self.key]
1183 )
1184 parent._configure_property(self.name, p, init=init, setparent=True)
1185 p._mapped_by_synonym = self.key
1187 self.parent = parent
1190class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
1191 """Declarative front-end for the :class:`.SynonymProperty` class.
1193 Public constructor is the :func:`_orm.synonym` function.
1195 .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
1196 compatible subclass for :class:`_orm.SynonymProperty`
1198 .. seealso::
1200 :ref:`synonyms` - Overview of synonyms
1202 """
1204 inherit_cache = True
1205 """:meta private:"""