Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/descriptor_props.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Descriptor properties are more "auxiliary" properties
9that exist as configurational elements, but don't participate
10as actively in the load/persist ORM loop.
12"""
13from __future__ import annotations
15from dataclasses import is_dataclass
16import inspect
17import itertools
18import operator
19import typing
20from typing import Any
21from typing import Callable
22from typing import Dict
23from typing import List
24from typing import NoReturn
25from typing import Optional
26from typing import Sequence
27from typing import Tuple
28from typing import Type
29from typing import TYPE_CHECKING
30from typing import TypeVar
31from typing import Union
32import weakref
34from . import attributes
35from . import util as orm_util
36from .base import _DeclarativeMapped
37from .base import LoaderCallableStatus
38from .base import Mapped
39from .base import PassiveFlag
40from .base import SQLORMOperations
41from .interfaces import _AttributeOptions
42from .interfaces import _IntrospectsAnnotations
43from .interfaces import _MapsColumns
44from .interfaces import MapperProperty
45from .interfaces import PropComparator
46from .util import _none_set
47from .util import de_stringify_annotation
48from .. import event
49from .. import exc as sa_exc
50from .. import schema
51from .. import sql
52from .. import util
53from ..sql import expression
54from ..sql import operators
55from ..sql.elements import BindParameter
56from ..util.typing import get_args
57from ..util.typing import is_fwd_ref
58from ..util.typing import is_pep593
61if typing.TYPE_CHECKING:
62 from ._typing import _InstanceDict
63 from ._typing import _RegistryType
64 from .attributes import History
65 from .attributes import InstrumentedAttribute
66 from .attributes import QueryableAttribute
67 from .context import ORMCompileState
68 from .decl_base import _ClassScanMapperConfig
69 from .mapper import Mapper
70 from .properties import ColumnProperty
71 from .properties import MappedColumn
72 from .state import InstanceState
73 from ..engine.base import Connection
74 from ..engine.row import Row
75 from ..sql._typing import _DMLColumnArgument
76 from ..sql._typing import _InfoType
77 from ..sql.elements import ClauseList
78 from ..sql.elements import ColumnElement
79 from ..sql.operators import OperatorType
80 from ..sql.schema import Column
81 from ..sql.selectable import Select
82 from ..util.typing import _AnnotationScanType
83 from ..util.typing import CallableReference
84 from ..util.typing import DescriptorReference
85 from ..util.typing import RODescriptorReference
87_T = TypeVar("_T", bound=Any)
88_PT = TypeVar("_PT", bound=Any)
91class DescriptorProperty(MapperProperty[_T]):
92 """:class:`.MapperProperty` which proxies access to a
93 user-defined descriptor."""
95 doc: Optional[str] = None
97 uses_objects = False
98 _links_to_entity = False
100 descriptor: DescriptorReference[Any]
102 def get_history(
103 self,
104 state: InstanceState[Any],
105 dict_: _InstanceDict,
106 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
107 ) -> History:
108 raise NotImplementedError()
110 def instrument_class(self, mapper: Mapper[Any]) -> None:
111 prop = self
113 class _ProxyImpl(attributes.AttributeImpl):
114 accepts_scalar_loader = False
115 load_on_unexpire = True
116 collection = False
118 @property
119 def uses_objects(self) -> bool: # type: ignore
120 return prop.uses_objects
122 def __init__(self, key: str):
123 self.key = key
125 def get_history(
126 self,
127 state: InstanceState[Any],
128 dict_: _InstanceDict,
129 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
130 ) -> History:
131 return prop.get_history(state, dict_, passive)
133 if self.descriptor is None:
134 desc = getattr(mapper.class_, self.key, None)
135 if mapper._is_userland_descriptor(self.key, desc):
136 self.descriptor = desc
138 if self.descriptor is None:
140 def fset(obj: Any, value: Any) -> None:
141 setattr(obj, self.name, value)
143 def fdel(obj: Any) -> None:
144 delattr(obj, self.name)
146 def fget(obj: Any) -> Any:
147 return getattr(obj, self.name)
149 self.descriptor = property(fget=fget, fset=fset, fdel=fdel)
151 proxy_attr = attributes.create_proxied_attribute(self.descriptor)(
152 self.parent.class_,
153 self.key,
154 self.descriptor,
155 lambda: self._comparator_factory(mapper),
156 doc=self.doc,
157 original_property=self,
158 )
159 proxy_attr.impl = _ProxyImpl(self.key)
160 mapper.class_manager.instrument_attribute(self.key, proxy_attr)
163_CompositeAttrType = Union[
164 str,
165 "Column[_T]",
166 "MappedColumn[_T]",
167 "InstrumentedAttribute[_T]",
168 "Mapped[_T]",
169]
172_CC = TypeVar("_CC", bound=Any)
175_composite_getters: weakref.WeakKeyDictionary[
176 Type[Any], Callable[[Any], Tuple[Any, ...]]
177] = weakref.WeakKeyDictionary()
180class CompositeProperty(
181 _MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC]
182):
183 """Defines a "composite" mapped attribute, representing a collection
184 of columns as one attribute.
186 :class:`.CompositeProperty` is constructed using the :func:`.composite`
187 function.
189 .. seealso::
191 :ref:`mapper_composite`
193 """
195 composite_class: Union[Type[_CC], Callable[..., _CC]]
196 attrs: Tuple[_CompositeAttrType[Any], ...]
198 _generated_composite_accessor: CallableReference[
199 Optional[Callable[[_CC], Tuple[Any, ...]]]
200 ]
202 comparator_factory: Type[Comparator[_CC]]
204 def __init__(
205 self,
206 _class_or_attr: Union[
207 None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any]
208 ] = None,
209 *attrs: _CompositeAttrType[Any],
210 attribute_options: Optional[_AttributeOptions] = None,
211 active_history: bool = False,
212 deferred: bool = False,
213 group: Optional[str] = None,
214 comparator_factory: Optional[Type[Comparator[_CC]]] = None,
215 info: Optional[_InfoType] = None,
216 **kwargs: Any,
217 ):
218 super().__init__(attribute_options=attribute_options)
220 if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)):
221 self.attrs = (_class_or_attr,) + attrs
222 # will initialize within declarative_scan
223 self.composite_class = None # type: ignore
224 else:
225 self.composite_class = _class_or_attr # type: ignore
226 self.attrs = attrs
228 self.active_history = active_history
229 self.deferred = deferred
230 self.group = group
231 self.comparator_factory = (
232 comparator_factory
233 if comparator_factory is not None
234 else self.__class__.Comparator
235 )
236 self._generated_composite_accessor = None
237 if info is not None:
238 self.info.update(info)
240 util.set_creation_order(self)
241 self._create_descriptor()
242 self._init_accessor()
244 def instrument_class(self, mapper: Mapper[Any]) -> None:
245 super().instrument_class(mapper)
246 self._setup_event_handlers()
248 def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]:
249 if self._generated_composite_accessor:
250 return self._generated_composite_accessor(value)
251 else:
252 try:
253 accessor = value.__composite_values__
254 except AttributeError as ae:
255 raise sa_exc.InvalidRequestError(
256 f"Composite class {self.composite_class.__name__} is not "
257 f"a dataclass and does not define a __composite_values__()"
258 " method; can't get state"
259 ) from ae
260 else:
261 return accessor() # type: ignore
263 def do_init(self) -> None:
264 """Initialization which occurs after the :class:`.Composite`
265 has been associated with its parent mapper.
267 """
268 self._setup_arguments_on_columns()
270 _COMPOSITE_FGET = object()
272 def _create_descriptor(self) -> None:
273 """Create the Python descriptor that will serve as
274 the access point on instances of the mapped class.
276 """
278 def fget(instance: Any) -> Any:
279 dict_ = attributes.instance_dict(instance)
280 state = attributes.instance_state(instance)
282 if self.key not in dict_:
283 # key not present. Iterate through related
284 # attributes, retrieve their values. This
285 # ensures they all load.
286 values = [
287 getattr(instance, key) for key in self._attribute_keys
288 ]
290 # current expected behavior here is that the composite is
291 # created on access if the object is persistent or if
292 # col attributes have non-None. This would be better
293 # if the composite were created unconditionally,
294 # but that would be a behavioral change.
295 if self.key not in dict_ and (
296 state.key is not None or not _none_set.issuperset(values)
297 ):
298 dict_[self.key] = self.composite_class(*values)
299 state.manager.dispatch.refresh(
300 state, self._COMPOSITE_FGET, [self.key]
301 )
303 return dict_.get(self.key, None)
305 def fset(instance: Any, value: Any) -> None:
306 dict_ = attributes.instance_dict(instance)
307 state = attributes.instance_state(instance)
308 attr = state.manager[self.key]
310 if attr.dispatch._active_history:
311 previous = fget(instance)
312 else:
313 previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE)
315 for fn in attr.dispatch.set:
316 value = fn(state, value, previous, attr.impl)
317 dict_[self.key] = value
318 if value is None:
319 for key in self._attribute_keys:
320 setattr(instance, key, None)
321 else:
322 for key, value in zip(
323 self._attribute_keys,
324 self._composite_values_from_instance(value),
325 ):
326 setattr(instance, key, value)
328 def fdel(instance: Any) -> None:
329 state = attributes.instance_state(instance)
330 dict_ = attributes.instance_dict(instance)
331 attr = state.manager[self.key]
333 if attr.dispatch._active_history:
334 previous = fget(instance)
335 dict_.pop(self.key, None)
336 else:
337 previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE)
339 attr = state.manager[self.key]
340 attr.dispatch.remove(state, previous, attr.impl)
341 for key in self._attribute_keys:
342 setattr(instance, key, None)
344 self.descriptor = property(fget, fset, fdel)
346 @util.preload_module("sqlalchemy.orm.properties")
347 def declarative_scan(
348 self,
349 decl_scan: _ClassScanMapperConfig,
350 registry: _RegistryType,
351 cls: Type[Any],
352 originating_module: Optional[str],
353 key: str,
354 mapped_container: Optional[Type[Mapped[Any]]],
355 annotation: Optional[_AnnotationScanType],
356 extracted_mapped_annotation: Optional[_AnnotationScanType],
357 is_dataclass_field: bool,
358 ) -> None:
359 MappedColumn = util.preloaded.orm_properties.MappedColumn
360 if (
361 self.composite_class is None
362 and extracted_mapped_annotation is None
363 ):
364 self._raise_for_required(key, cls)
365 argument = extracted_mapped_annotation
367 if is_pep593(argument):
368 argument = get_args(argument)[0]
370 if argument and self.composite_class is None:
371 if isinstance(argument, str) or is_fwd_ref(
372 argument, check_generic=True
373 ):
374 if originating_module is None:
375 str_arg = (
376 argument.__forward_arg__
377 if hasattr(argument, "__forward_arg__")
378 else str(argument)
379 )
380 raise sa_exc.ArgumentError(
381 f"Can't use forward ref {argument} for composite "
382 f"class argument; set up the type as Mapped[{str_arg}]"
383 )
384 argument = de_stringify_annotation(
385 cls, argument, originating_module, include_generic=True
386 )
388 self.composite_class = argument
390 if is_dataclass(self.composite_class):
391 self._setup_for_dataclass(registry, cls, originating_module, key)
392 else:
393 for attr in self.attrs:
394 if (
395 isinstance(attr, (MappedColumn, schema.Column))
396 and attr.name is None
397 ):
398 raise sa_exc.ArgumentError(
399 "Composite class column arguments must be named "
400 "unless a dataclass is used"
401 )
402 self._init_accessor()
404 def _init_accessor(self) -> None:
405 if is_dataclass(self.composite_class) and not hasattr(
406 self.composite_class, "__composite_values__"
407 ):
408 insp = inspect.signature(self.composite_class)
409 getter = operator.attrgetter(
410 *[p.name for p in insp.parameters.values()]
411 )
412 if len(insp.parameters) == 1:
413 self._generated_composite_accessor = lambda obj: (getter(obj),)
414 else:
415 self._generated_composite_accessor = getter
417 if (
418 self.composite_class is not None
419 and isinstance(self.composite_class, type)
420 and self.composite_class not in _composite_getters
421 ):
422 if self._generated_composite_accessor is not None:
423 _composite_getters[self.composite_class] = (
424 self._generated_composite_accessor
425 )
426 elif hasattr(self.composite_class, "__composite_values__"):
427 _composite_getters[self.composite_class] = (
428 lambda obj: obj.__composite_values__()
429 )
431 @util.preload_module("sqlalchemy.orm.properties")
432 @util.preload_module("sqlalchemy.orm.decl_base")
433 def _setup_for_dataclass(
434 self,
435 registry: _RegistryType,
436 cls: Type[Any],
437 originating_module: Optional[str],
438 key: str,
439 ) -> None:
440 MappedColumn = util.preloaded.orm_properties.MappedColumn
442 decl_base = util.preloaded.orm_decl_base
444 insp = inspect.signature(self.composite_class)
445 for param, attr in itertools.zip_longest(
446 insp.parameters.values(), self.attrs
447 ):
448 if param is None:
449 raise sa_exc.ArgumentError(
450 f"number of composite attributes "
451 f"{len(self.attrs)} exceeds "
452 f"that of the number of attributes in class "
453 f"{self.composite_class.__name__} {len(insp.parameters)}"
454 )
455 if attr is None:
456 # fill in missing attr spots with empty MappedColumn
457 attr = MappedColumn()
458 self.attrs += (attr,)
460 if isinstance(attr, MappedColumn):
461 attr.declarative_scan_for_composite(
462 registry,
463 cls,
464 originating_module,
465 key,
466 param.name,
467 param.annotation,
468 )
469 elif isinstance(attr, schema.Column):
470 decl_base._undefer_column_name(param.name, attr)
472 @util.memoized_property
473 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
474 return [getattr(self.parent.class_, prop.key) for prop in self.props]
476 @util.memoized_property
477 @util.preload_module("orm.properties")
478 def props(self) -> Sequence[MapperProperty[Any]]:
479 props = []
480 MappedColumn = util.preloaded.orm_properties.MappedColumn
482 for attr in self.attrs:
483 if isinstance(attr, str):
484 prop = self.parent.get_property(attr, _configure_mappers=False)
485 elif isinstance(attr, schema.Column):
486 prop = self.parent._columntoproperty[attr]
487 elif isinstance(attr, MappedColumn):
488 prop = self.parent._columntoproperty[attr.column]
489 elif isinstance(attr, attributes.InstrumentedAttribute):
490 prop = attr.property
491 else:
492 prop = None
494 if not isinstance(prop, MapperProperty):
495 raise sa_exc.ArgumentError(
496 "Composite expects Column objects or mapped "
497 f"attributes/attribute names as arguments, got: {attr!r}"
498 )
500 props.append(prop)
501 return props
503 @util.non_memoized_property
504 @util.preload_module("orm.properties")
505 def columns(self) -> Sequence[Column[Any]]:
506 MappedColumn = util.preloaded.orm_properties.MappedColumn
507 return [
508 a.column if isinstance(a, MappedColumn) else a
509 for a in self.attrs
510 if isinstance(a, (schema.Column, MappedColumn))
511 ]
513 @property
514 def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]:
515 return self
517 @property
518 def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
519 return [(c, 0) for c in self.columns if c.table is None]
521 @util.preload_module("orm.properties")
522 def _setup_arguments_on_columns(self) -> None:
523 """Propagate configuration arguments made on this composite
524 to the target columns, for those that apply.
526 """
527 ColumnProperty = util.preloaded.orm_properties.ColumnProperty
529 for prop in self.props:
530 if not isinstance(prop, ColumnProperty):
531 continue
532 else:
533 cprop = prop
535 cprop.active_history = self.active_history
536 if self.deferred:
537 cprop.deferred = self.deferred
538 cprop.strategy_key = (("deferred", True), ("instrument", True))
539 cprop.group = self.group
541 def _setup_event_handlers(self) -> None:
542 """Establish events that populate/expire the composite attribute."""
544 def load_handler(
545 state: InstanceState[Any], context: ORMCompileState
546 ) -> None:
547 _load_refresh_handler(state, context, None, is_refresh=False)
549 def refresh_handler(
550 state: InstanceState[Any],
551 context: ORMCompileState,
552 to_load: Optional[Sequence[str]],
553 ) -> None:
554 # note this corresponds to sqlalchemy.ext.mutable load_attrs()
556 if not to_load or (
557 {self.key}.union(self._attribute_keys)
558 ).intersection(to_load):
559 _load_refresh_handler(state, context, to_load, is_refresh=True)
561 def _load_refresh_handler(
562 state: InstanceState[Any],
563 context: ORMCompileState,
564 to_load: Optional[Sequence[str]],
565 is_refresh: bool,
566 ) -> None:
567 dict_ = state.dict
569 # if context indicates we are coming from the
570 # fget() handler, this already set the value; skip the
571 # handler here. (other handlers like mutablecomposite will still
572 # want to catch it)
573 # there's an insufficiency here in that the fget() handler
574 # really should not be using the refresh event and there should
575 # be some other event that mutablecomposite can subscribe
576 # towards for this.
578 if (
579 not is_refresh or context is self._COMPOSITE_FGET
580 ) and self.key in dict_:
581 return
583 # if column elements aren't loaded, skip.
584 # __get__() will initiate a load for those
585 # columns
586 for k in self._attribute_keys:
587 if k not in dict_:
588 return
590 dict_[self.key] = self.composite_class(
591 *[state.dict[key] for key in self._attribute_keys]
592 )
594 def expire_handler(
595 state: InstanceState[Any], keys: Optional[Sequence[str]]
596 ) -> None:
597 if keys is None or set(self._attribute_keys).intersection(keys):
598 state.dict.pop(self.key, None)
600 def insert_update_handler(
601 mapper: Mapper[Any],
602 connection: Connection,
603 state: InstanceState[Any],
604 ) -> None:
605 """After an insert or update, some columns may be expired due
606 to server side defaults, or re-populated due to client side
607 defaults. Pop out the composite value here so that it
608 recreates.
610 """
612 state.dict.pop(self.key, None)
614 event.listen(
615 self.parent, "after_insert", insert_update_handler, raw=True
616 )
617 event.listen(
618 self.parent, "after_update", insert_update_handler, raw=True
619 )
620 event.listen(
621 self.parent, "load", load_handler, raw=True, propagate=True
622 )
623 event.listen(
624 self.parent, "refresh", refresh_handler, raw=True, propagate=True
625 )
626 event.listen(
627 self.parent, "expire", expire_handler, raw=True, propagate=True
628 )
630 proxy_attr = self.parent.class_manager[self.key]
631 proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore
632 proxy_attr.impl.dispatch._active_history = self.active_history
634 # TODO: need a deserialize hook here
636 @util.memoized_property
637 def _attribute_keys(self) -> Sequence[str]:
638 return [prop.key for prop in self.props]
640 def _populate_composite_bulk_save_mappings_fn(
641 self,
642 ) -> Callable[[Dict[str, Any]], None]:
643 if self._generated_composite_accessor:
644 get_values = self._generated_composite_accessor
645 else:
647 def get_values(val: Any) -> Tuple[Any]:
648 return val.__composite_values__() # type: ignore
650 attrs = [prop.key for prop in self.props]
652 def populate(dest_dict: Dict[str, Any]) -> None:
653 dest_dict.update(
654 {
655 key: val
656 for key, val in zip(
657 attrs, get_values(dest_dict.pop(self.key))
658 )
659 }
660 )
662 return populate
664 def get_history(
665 self,
666 state: InstanceState[Any],
667 dict_: _InstanceDict,
668 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
669 ) -> History:
670 """Provided for userland code that uses attributes.get_history()."""
672 added: List[Any] = []
673 deleted: List[Any] = []
675 has_history = False
676 for prop in self.props:
677 key = prop.key
678 hist = state.manager[key].impl.get_history(state, dict_)
679 if hist.has_changes():
680 has_history = True
682 non_deleted = hist.non_deleted()
683 if non_deleted:
684 added.extend(non_deleted)
685 else:
686 added.append(None)
687 if hist.deleted:
688 deleted.extend(hist.deleted)
689 else:
690 deleted.append(None)
692 if has_history:
693 return attributes.History(
694 [self.composite_class(*added)],
695 (),
696 [self.composite_class(*deleted)],
697 )
698 else:
699 return attributes.History((), [self.composite_class(*added)], ())
701 def _comparator_factory(
702 self, mapper: Mapper[Any]
703 ) -> Composite.Comparator[_CC]:
704 return self.comparator_factory(self, mapper)
706 class CompositeBundle(orm_util.Bundle[_T]):
707 def __init__(
708 self,
709 property_: Composite[_T],
710 expr: ClauseList,
711 ):
712 self.property = property_
713 super().__init__(property_.key, *expr)
715 def create_row_processor(
716 self,
717 query: Select[Any],
718 procs: Sequence[Callable[[Row[Any]], Any]],
719 labels: Sequence[str],
720 ) -> Callable[[Row[Any]], Any]:
721 def proc(row: Row[Any]) -> Any:
722 return self.property.composite_class(
723 *[proc(row) for proc in procs]
724 )
726 return proc
728 class Comparator(PropComparator[_PT]):
729 """Produce boolean, comparison, and other operators for
730 :class:`.Composite` attributes.
732 See the example in :ref:`composite_operations` for an overview
733 of usage , as well as the documentation for :class:`.PropComparator`.
735 .. seealso::
737 :class:`.PropComparator`
739 :class:`.ColumnOperators`
741 :ref:`types_operators`
743 :attr:`.TypeEngine.comparator_factory`
745 """
747 # https://github.com/python/mypy/issues/4266
748 __hash__ = None # type: ignore
750 prop: RODescriptorReference[Composite[_PT]]
752 @util.memoized_property
753 def clauses(self) -> ClauseList:
754 return expression.ClauseList(
755 group=False, *self._comparable_elements
756 )
758 def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]:
759 return self.expression
761 @util.memoized_property
762 def expression(self) -> CompositeProperty.CompositeBundle[_PT]:
763 clauses = self.clauses._annotate(
764 {
765 "parententity": self._parententity,
766 "parentmapper": self._parententity,
767 "proxy_key": self.prop.key,
768 }
769 )
770 return CompositeProperty.CompositeBundle(self.prop, clauses)
772 def _bulk_update_tuples(
773 self, value: Any
774 ) -> Sequence[Tuple[_DMLColumnArgument, Any]]:
775 if isinstance(value, BindParameter):
776 value = value.value
778 values: Sequence[Any]
780 if value is None:
781 values = [None for key in self.prop._attribute_keys]
782 elif isinstance(self.prop.composite_class, type) and isinstance(
783 value, self.prop.composite_class
784 ):
785 values = self.prop._composite_values_from_instance(
786 value # type: ignore[arg-type]
787 )
788 else:
789 raise sa_exc.ArgumentError(
790 "Can't UPDATE composite attribute %s to %r"
791 % (self.prop, value)
792 )
794 return list(zip(self._comparable_elements, values))
796 @util.memoized_property
797 def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]:
798 if self._adapt_to_entity:
799 return [
800 getattr(self._adapt_to_entity.entity, prop.key)
801 for prop in self.prop._comparable_elements
802 ]
803 else:
804 return self.prop._comparable_elements
806 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
807 return self._compare(operators.eq, other)
809 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
810 return self._compare(operators.ne, other)
812 def __lt__(self, other: Any) -> ColumnElement[bool]:
813 return self._compare(operators.lt, other)
815 def __gt__(self, other: Any) -> ColumnElement[bool]:
816 return self._compare(operators.gt, other)
818 def __le__(self, other: Any) -> ColumnElement[bool]:
819 return self._compare(operators.le, other)
821 def __ge__(self, other: Any) -> ColumnElement[bool]:
822 return self._compare(operators.ge, other)
824 # what might be interesting would be if we create
825 # an instance of the composite class itself with
826 # the columns as data members, then use "hybrid style" comparison
827 # to create these comparisons. then your Point.__eq__() method could
828 # be where comparison behavior is defined for SQL also. Likely
829 # not a good choice for default behavior though, not clear how it would
830 # work w/ dataclasses, etc. also no demand for any of this anyway.
831 def _compare(
832 self, operator: OperatorType, other: Any
833 ) -> ColumnElement[bool]:
834 values: Sequence[Any]
835 if other is None:
836 values = [None] * len(self.prop._comparable_elements)
837 else:
838 values = self.prop._composite_values_from_instance(other)
839 comparisons = [
840 operator(a, b)
841 for a, b in zip(self.prop._comparable_elements, values)
842 ]
843 if self._adapt_to_entity:
844 assert self.adapter is not None
845 comparisons = [self.adapter(x) for x in comparisons]
846 return sql.and_(*comparisons)
848 def __str__(self) -> str:
849 return str(self.parent.class_.__name__) + "." + self.key
852class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]):
853 """Declarative-compatible front-end for the :class:`.CompositeProperty`
854 class.
856 Public constructor is the :func:`_orm.composite` function.
858 .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative
859 compatible subclass of :class:`_orm.CompositeProperty`.
861 .. seealso::
863 :ref:`mapper_composite`
865 """
867 inherit_cache = True
868 """:meta private:"""
871class ConcreteInheritedProperty(DescriptorProperty[_T]):
872 """A 'do nothing' :class:`.MapperProperty` that disables
873 an attribute on a concrete subclass that is only present
874 on the inherited mapper, not the concrete classes' mapper.
876 Cases where this occurs include:
878 * When the superclass mapper is mapped against a
879 "polymorphic union", which includes all attributes from
880 all subclasses.
881 * When a relationship() is configured on an inherited mapper,
882 but not on the subclass mapper. Concrete mappers require
883 that relationship() is configured explicitly on each
884 subclass.
886 """
888 def _comparator_factory(
889 self, mapper: Mapper[Any]
890 ) -> Type[PropComparator[_T]]:
891 comparator_callable = None
893 for m in self.parent.iterate_to_root():
894 p = m._props[self.key]
895 if getattr(p, "comparator_factory", None) is not None:
896 comparator_callable = p.comparator_factory
897 break
898 assert comparator_callable is not None
899 return comparator_callable(p, mapper) # type: ignore
901 def __init__(self) -> None:
902 super().__init__()
904 def warn() -> NoReturn:
905 raise AttributeError(
906 "Concrete %s does not implement "
907 "attribute %r at the instance level. Add "
908 "this property explicitly to %s."
909 % (self.parent, self.key, self.parent)
910 )
912 class NoninheritedConcreteProp:
913 def __set__(s: Any, obj: Any, value: Any) -> NoReturn:
914 warn()
916 def __delete__(s: Any, obj: Any) -> NoReturn:
917 warn()
919 def __get__(s: Any, obj: Any, owner: Any) -> Any:
920 if obj is None:
921 return self.descriptor
922 warn()
924 self.descriptor = NoninheritedConcreteProp()
927class SynonymProperty(DescriptorProperty[_T]):
928 """Denote an attribute name as a synonym to a mapped property,
929 in that the attribute will mirror the value and expression behavior
930 of another attribute.
932 :class:`.Synonym` is constructed using the :func:`_orm.synonym`
933 function.
935 .. seealso::
937 :ref:`synonyms` - Overview of synonyms
939 """
941 comparator_factory: Optional[Type[PropComparator[_T]]]
943 def __init__(
944 self,
945 name: str,
946 map_column: Optional[bool] = None,
947 descriptor: Optional[Any] = None,
948 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
949 attribute_options: Optional[_AttributeOptions] = None,
950 info: Optional[_InfoType] = None,
951 doc: Optional[str] = None,
952 ):
953 super().__init__(attribute_options=attribute_options)
955 self.name = name
956 self.map_column = map_column
957 self.descriptor = descriptor
958 self.comparator_factory = comparator_factory
959 if doc:
960 self.doc = doc
961 elif descriptor and descriptor.__doc__:
962 self.doc = descriptor.__doc__
963 else:
964 self.doc = None
965 if info:
966 self.info.update(info)
968 util.set_creation_order(self)
970 if not TYPE_CHECKING:
972 @property
973 def uses_objects(self) -> bool:
974 return getattr(self.parent.class_, self.name).impl.uses_objects
976 # TODO: when initialized, check _proxied_object,
977 # emit a warning if its not a column-based property
979 @util.memoized_property
980 def _proxied_object(
981 self,
982 ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]:
983 attr = getattr(self.parent.class_, self.name)
984 if not hasattr(attr, "property") or not isinstance(
985 attr.property, MapperProperty
986 ):
987 # attribute is a non-MapperProprerty proxy such as
988 # hybrid or association proxy
989 if isinstance(attr, attributes.QueryableAttribute):
990 return attr.comparator
991 elif isinstance(attr, SQLORMOperations):
992 # assocaition proxy comes here
993 return attr
995 raise sa_exc.InvalidRequestError(
996 """synonym() attribute "%s.%s" only supports """
997 """ORM mapped attributes, got %r"""
998 % (self.parent.class_.__name__, self.name, attr)
999 )
1000 return attr.property
1002 def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]:
1003 prop = self._proxied_object
1005 if isinstance(prop, MapperProperty):
1006 if self.comparator_factory:
1007 comp = self.comparator_factory(prop, mapper)
1008 else:
1009 comp = prop.comparator_factory(prop, mapper)
1010 return comp
1011 else:
1012 return prop
1014 def get_history(
1015 self,
1016 state: InstanceState[Any],
1017 dict_: _InstanceDict,
1018 passive: PassiveFlag = PassiveFlag.PASSIVE_OFF,
1019 ) -> History:
1020 attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name)
1021 return attr.impl.get_history(state, dict_, passive=passive)
1023 @util.preload_module("sqlalchemy.orm.properties")
1024 def set_parent(self, parent: Mapper[Any], init: bool) -> None:
1025 properties = util.preloaded.orm_properties
1027 if self.map_column:
1028 # implement the 'map_column' option.
1029 if self.key not in parent.persist_selectable.c:
1030 raise sa_exc.ArgumentError(
1031 "Can't compile synonym '%s': no column on table "
1032 "'%s' named '%s'"
1033 % (
1034 self.name,
1035 parent.persist_selectable.description,
1036 self.key,
1037 )
1038 )
1039 elif (
1040 parent.persist_selectable.c[self.key]
1041 in parent._columntoproperty
1042 and parent._columntoproperty[
1043 parent.persist_selectable.c[self.key]
1044 ].key
1045 == self.name
1046 ):
1047 raise sa_exc.ArgumentError(
1048 "Can't call map_column=True for synonym %r=%r, "
1049 "a ColumnProperty already exists keyed to the name "
1050 "%r for column %r"
1051 % (self.key, self.name, self.name, self.key)
1052 )
1053 p: ColumnProperty[Any] = properties.ColumnProperty(
1054 parent.persist_selectable.c[self.key]
1055 )
1056 parent._configure_property(self.name, p, init=init, setparent=True)
1057 p._mapped_by_synonym = self.key
1059 self.parent = parent
1062class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]):
1063 """Declarative front-end for the :class:`.SynonymProperty` class.
1065 Public constructor is the :func:`_orm.synonym` function.
1067 .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative
1068 compatible subclass for :class:`_orm.SynonymProperty`
1070 .. seealso::
1072 :ref:`synonyms` - Overview of synonyms
1074 """
1076 inherit_cache = True
1077 """:meta private:"""