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