# orm/descriptor_props.py # Copyright (C) 2005-2026 the SQLAlchemy authors and contributors # # # This module is part of SQLAlchemy and is released under # the MIT License: https://www.opensource.org/licenses/mit-license.php """Descriptor properties are more "auxiliary" properties that exist as configurational elements, but don't participate as actively in the load/persist ORM loop. """ from __future__ import annotations from dataclasses import is_dataclass import inspect import itertools import operator import typing from typing import Any from typing import Callable from typing import Dict from typing import get_args from typing import List from typing import NoReturn from typing import Optional from typing import Sequence from typing import Tuple from typing import Type from typing import TYPE_CHECKING from typing import TypeVar from typing import Union import weakref from . import attributes from . import util as orm_util from .base import _DeclarativeMapped from .base import DONT_SET from .base import LoaderCallableStatus from .base import Mapped from .base import PassiveFlag from .base import SQLORMOperations from .interfaces import _AttributeOptions from .interfaces import _IntrospectsAnnotations from .interfaces import _MapsColumns from .interfaces import MapperProperty from .interfaces import PropComparator from .util import de_stringify_annotation from .. import event from .. import exc as sa_exc from .. import schema from .. import sql from .. import util from ..sql import expression from ..sql import operators from ..sql.base import _NoArg from ..sql.elements import BindParameter from ..util.typing import de_optionalize_union_types from ..util.typing import includes_none from ..util.typing import is_fwd_ref from ..util.typing import is_pep593 from ..util.typing import is_union from ..util.typing import TupleAny from ..util.typing import Unpack if typing.TYPE_CHECKING: from ._typing import _InstanceDict from ._typing import _RegistryType from .attributes import History from .attributes import InstrumentedAttribute from .attributes import QueryableAttribute from .context import _ORMCompileState from .decl_base import _ClassScanAbstractConfig from .decl_base import _DeclarativeMapperConfig from .interfaces import _DataclassArguments from .mapper import Mapper from .properties import ColumnProperty from .properties import MappedColumn from .state import InstanceState from ..engine.base import Connection from ..engine.row import Row from ..sql._typing import _DMLColumnArgument from ..sql._typing import _InfoType from ..sql.elements import ClauseList from ..sql.elements import ColumnElement from ..sql.operators import OperatorType from ..sql.schema import Column from ..sql.selectable import Select from ..util.typing import _AnnotationScanType from ..util.typing import CallableReference from ..util.typing import DescriptorReference from ..util.typing import RODescriptorReference _T = TypeVar("_T", bound=Any) _PT = TypeVar("_PT", bound=Any) class DescriptorProperty(MapperProperty[_T]): """:class:`.MapperProperty` which proxies access to a user-defined descriptor.""" doc: Optional[str] = None uses_objects = False _links_to_entity = False descriptor: DescriptorReference[Any] def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]: raise NotImplementedError( "This MapperProperty does not implement column loader strategies" ) def get_history( self, state: InstanceState[Any], dict_: _InstanceDict, passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, ) -> History: raise NotImplementedError() def instrument_class(self, mapper: Mapper[Any]) -> None: prop = self class _ProxyImpl(attributes._AttributeImpl): accepts_scalar_loader = False load_on_unexpire = True collection = False @property def uses_objects(self) -> bool: # type: ignore[override] return prop.uses_objects def __init__(self, key: str): self.key = key def get_history( self, state: InstanceState[Any], dict_: _InstanceDict, passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, ) -> History: return prop.get_history(state, dict_, passive) if self.descriptor is None: desc = getattr(mapper.class_, self.key, None) if mapper._is_userland_descriptor(self.key, desc): self.descriptor = desc if self.descriptor is None: def fset(obj: Any, value: Any) -> None: setattr(obj, self.name, value) def fdel(obj: Any) -> None: delattr(obj, self.name) def fget(obj: Any) -> Any: return getattr(obj, self.name) self.descriptor = property(fget=fget, fset=fset, fdel=fdel) proxy_attr = attributes._create_proxied_attribute(self.descriptor)( self.parent.class_, self.key, self.descriptor, lambda: self._comparator_factory(mapper), doc=self.doc, original_property=self, ) proxy_attr.impl = _ProxyImpl(self.key) mapper.class_manager.instrument_attribute(self.key, proxy_attr) _CompositeAttrType = Union[ str, "Column[_T]", "MappedColumn[_T]", "InstrumentedAttribute[_T]", "Mapped[_T]", ] _CC = TypeVar("_CC", bound=Any) _composite_getters: weakref.WeakKeyDictionary[ Type[Any], Callable[[Any], Tuple[Any, ...]] ] = weakref.WeakKeyDictionary() class CompositeProperty( _MapsColumns[_CC], _IntrospectsAnnotations, DescriptorProperty[_CC] ): """Defines a "composite" mapped attribute, representing a collection of columns as one attribute. :class:`.CompositeProperty` is constructed using the :func:`.composite` function. .. seealso:: :ref:`mapper_composite` """ composite_class: Union[Type[_CC], Callable[..., _CC]] attrs: Tuple[_CompositeAttrType[Any], ...] column_template: Optional[str] _generated_composite_accessor: CallableReference[ Optional[Callable[[_CC], Tuple[Any, ...]]] ] comparator_factory: Type[Comparator[_CC]] def __init__( self, _class_or_attr: Union[ None, Type[_CC], Callable[..., _CC], _CompositeAttrType[Any] ] = None, *attrs: _CompositeAttrType[Any], return_none_on: Union[ _NoArg, None, Callable[..., bool] ] = _NoArg.NO_ARG, attribute_options: Optional[_AttributeOptions] = None, active_history: bool = False, deferred: bool = False, group: Optional[str] = None, comparator_factory: Optional[Type[Comparator[_CC]]] = None, column_template: Optional[str] = None, info: Optional[_InfoType] = None, **kwargs: Any, ): super().__init__(attribute_options=attribute_options) if isinstance(_class_or_attr, (Mapped, str, sql.ColumnElement)): self.attrs = (_class_or_attr,) + attrs # will initialize within declarative_scan self.composite_class = None # type: ignore[assignment] else: self.composite_class = _class_or_attr # type: ignore[assignment] self.attrs = attrs if column_template is not None: try: column_template % "x" except (TypeError, ValueError) as te: raise sa_exc.ArgumentError( f"column_template {column_template!r} is not a valid " "template; expected a string containing exactly one " "'%s' placeholder" ) from te self.column_template = column_template self.return_none_on = return_none_on self.active_history = active_history self.deferred = deferred self.group = group self.comparator_factory = ( comparator_factory if comparator_factory is not None else self.__class__.Comparator ) self._generated_composite_accessor = None if info is not None: self.info.update(info) util.set_creation_order(self) self._create_descriptor() self._init_accessor() @util.memoized_property def _construct_composite(self) -> Callable[..., Any]: return_none_on = self.return_none_on if callable(return_none_on): def construct(*args: Any) -> Any: if return_none_on(*args): return None else: return self.composite_class(*args) return construct else: return self.composite_class def instrument_class(self, mapper: Mapper[Any]) -> None: super().instrument_class(mapper) self._setup_event_handlers() def _composite_values_from_instance(self, value: _CC) -> Tuple[Any, ...]: if self._generated_composite_accessor: return self._generated_composite_accessor(value) else: try: accessor = value.__composite_values__ except AttributeError as ae: raise sa_exc.InvalidRequestError( f"Composite class {self.composite_class.__name__} is not " f"a dataclass and does not define a __composite_values__()" " method; can't get state" ) from ae else: return accessor() # type: ignore[no-any-return] def do_init(self) -> None: """Initialization which occurs after the :class:`.Composite` has been associated with its parent mapper. """ self._setup_arguments_on_columns() _COMPOSITE_FGET = object() def _create_descriptor(self) -> None: """Create the Python descriptor that will serve as the access point on instances of the mapped class. """ def fget(instance: Any) -> Any: dict_ = attributes.instance_dict(instance) state = attributes.instance_state(instance) if self.key not in dict_: # key not present. Iterate through related # attributes, retrieve their values. This # ensures they all load. values = [ getattr(instance, key) for key in self._attribute_keys ] if self.key not in dict_: dict_[self.key] = self._construct_composite(*values) state.manager.dispatch.refresh( state, self._COMPOSITE_FGET, [self.key] ) return dict_.get(self.key, None) def fset(instance: Any, value: Any) -> None: if value is LoaderCallableStatus.DONT_SET: return dict_ = attributes.instance_dict(instance) state = attributes.instance_state(instance) attr = state.manager[self.key] if attr.dispatch._active_history: previous = fget(instance) else: previous = dict_.get(self.key, LoaderCallableStatus.NO_VALUE) for fn in attr.dispatch.set: value = fn(state, value, previous, attr.impl) dict_[self.key] = value if value is None: for key in self._attribute_keys: setattr(instance, key, None) else: for key, value in zip( self._attribute_keys, self._composite_values_from_instance(value), ): setattr(instance, key, value) def fdel(instance: Any) -> None: state = attributes.instance_state(instance) dict_ = attributes.instance_dict(instance) attr = state.manager[self.key] if attr.dispatch._active_history: previous = fget(instance) dict_.pop(self.key, None) else: previous = dict_.pop(self.key, LoaderCallableStatus.NO_VALUE) attr = state.manager[self.key] attr.dispatch.remove(state, previous, attr.impl) for key in self._attribute_keys: setattr(instance, key, None) self.descriptor = property(fget, fset, fdel) @util.preload_module("sqlalchemy.orm.properties") def declarative_scan( self, decl_scan: _DeclarativeMapperConfig, registry: _RegistryType, cls: Type[Any], originating_module: Optional[str], key: str, mapped_container: Optional[Type[Mapped[Any]]], annotation: Optional[_AnnotationScanType], extracted_mapped_annotation: Optional[_AnnotationScanType], is_dataclass_field: bool, ) -> None: MappedColumn = util.preloaded.orm_properties.MappedColumn if ( self.composite_class is None and extracted_mapped_annotation is None ): self._raise_for_required(key, cls) argument = extracted_mapped_annotation if is_pep593(argument): argument = get_args(argument)[0] if argument and self.composite_class is None: if isinstance(argument, str) or is_fwd_ref( argument, check_generic=True ): if originating_module is None: str_arg = ( argument.__forward_arg__ if hasattr(argument, "__forward_arg__") else str(argument) ) raise sa_exc.ArgumentError( f"Can't use forward ref {argument} for composite " f"class argument; set up the type as Mapped[{str_arg}]" ) argument = de_stringify_annotation( cls, argument, originating_module, include_generic=True ) if is_union(argument) and includes_none(argument): if self.return_none_on is _NoArg.NO_ARG: self.return_none_on = lambda *args: all( arg is None for arg in args ) argument = de_optionalize_union_types(argument) self.composite_class = argument if is_dataclass(self.composite_class): self._setup_for_dataclass( decl_scan, registry, cls, originating_module, key ) else: if self.column_template is not None: raise sa_exc.ArgumentError( "column_template is only supported when composite_class " "is a dataclass" ) for attr in self.attrs: if ( isinstance(attr, (MappedColumn, schema.Column)) and attr.name is None ): raise sa_exc.ArgumentError( "Composite class column arguments must be named " "unless a dataclass is used" ) self._init_accessor() def _init_accessor(self) -> None: if is_dataclass(self.composite_class) and not hasattr( self.composite_class, "__composite_values__" ): insp = inspect.signature(self.composite_class) getter = operator.attrgetter( *[p.name for p in insp.parameters.values()] ) if len(insp.parameters) == 1: self._generated_composite_accessor = lambda obj: (getter(obj),) else: self._generated_composite_accessor = getter if ( self.composite_class is not None and isinstance(self.composite_class, type) and self.composite_class not in _composite_getters ): if self._generated_composite_accessor is not None: _composite_getters[self.composite_class] = ( self._generated_composite_accessor ) elif hasattr(self.composite_class, "__composite_values__"): _composite_getters[self.composite_class] = ( lambda obj: obj.__composite_values__() ) @util.preload_module("sqlalchemy.orm.properties") @util.preload_module("sqlalchemy.orm.decl_base") def _setup_for_dataclass( self, decl_scan: _DeclarativeMapperConfig, registry: _RegistryType, cls: Type[Any], originating_module: Optional[str], key: str, ) -> None: MappedColumn = util.preloaded.orm_properties.MappedColumn decl_base = util.preloaded.orm_decl_base insp = inspect.signature(self.composite_class) for param, attr in itertools.zip_longest( insp.parameters.values(), self.attrs ): if param is None: raise sa_exc.ArgumentError( f"number of composite attributes " f"{len(self.attrs)} exceeds " f"that of the number of attributes in class " f"{self.composite_class.__name__} {len(insp.parameters)}" ) if attr is None: # fill in missing attr spots with empty MappedColumn, # or one named from column_template if present if self.column_template is not None: attr = MappedColumn(self.column_template % param.name) else: attr = MappedColumn() self.attrs += (attr,) if isinstance(attr, MappedColumn): attr.declarative_scan_for_composite( decl_scan, registry, cls, originating_module, key, param.name, param.annotation, ) elif isinstance(attr, schema.Column): decl_base._undefer_column_name(param.name, attr) @util.memoized_property def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]: return [getattr(self.parent.class_, prop.key) for prop in self.props] @util.memoized_property @util.preload_module("orm.properties") def props(self) -> Sequence[MapperProperty[Any]]: props = [] MappedColumn = util.preloaded.orm_properties.MappedColumn for attr in self.attrs: if isinstance(attr, str): prop = self.parent.get_property(attr, _configure_mappers=False) elif isinstance(attr, schema.Column): prop = self.parent._columntoproperty[attr] elif isinstance(attr, MappedColumn): prop = self.parent._columntoproperty[attr.column] elif isinstance(attr, attributes.InstrumentedAttribute): prop = attr.property else: prop = None if not isinstance(prop, MapperProperty): raise sa_exc.ArgumentError( "Composite expects Column objects or mapped " f"attributes/attribute names as arguments, got: {attr!r}" ) props.append(prop) return props def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]: return self._comparable_elements @util.non_memoized_property @util.preload_module("orm.properties") def columns(self) -> Sequence[Column[Any]]: MappedColumn = util.preloaded.orm_properties.MappedColumn return [ a.column if isinstance(a, MappedColumn) else a for a in self.attrs if isinstance(a, (schema.Column, MappedColumn)) ] @property def mapper_property_to_assign(self) -> Optional[MapperProperty[_CC]]: return self @property def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]: return [(c, 0) for c in self.columns if c.table is None] @util.preload_module("orm.properties") def _setup_arguments_on_columns(self) -> None: """Propagate configuration arguments made on this composite to the target columns, for those that apply. """ ColumnProperty = util.preloaded.orm_properties.ColumnProperty for prop in self.props: if not isinstance(prop, ColumnProperty): continue else: cprop = prop cprop.active_history = self.active_history if self.deferred: cprop.deferred = self.deferred cprop.strategy_key = (("deferred", True), ("instrument", True)) cprop.group = self.group def _setup_event_handlers(self) -> None: """Establish events that populate/expire the composite attribute.""" def load_handler( state: InstanceState[Any], context: _ORMCompileState ) -> None: _load_refresh_handler(state, context, None, is_refresh=False) def refresh_handler( state: InstanceState[Any], context: _ORMCompileState, to_load: Optional[Sequence[str]], ) -> None: # note this corresponds to sqlalchemy.ext.mutable load_attrs() if not to_load or ( {self.key}.union(self._attribute_keys) ).intersection(to_load): _load_refresh_handler(state, context, to_load, is_refresh=True) def _load_refresh_handler( state: InstanceState[Any], context: _ORMCompileState, to_load: Optional[Sequence[str]], is_refresh: bool, ) -> None: dict_ = state.dict # if context indicates we are coming from the # fget() handler, this already set the value; skip the # handler here. (other handlers like mutablecomposite will still # want to catch it) # there's an insufficiency here in that the fget() handler # really should not be using the refresh event and there should # be some other event that mutablecomposite can subscribe # towards for this. if ( not is_refresh or context is self._COMPOSITE_FGET ) and self.key in dict_: return # if column elements aren't loaded, skip. # __get__() will initiate a load for those # columns for k in self._attribute_keys: if k not in dict_: return dict_[self.key] = self._construct_composite( *[state.dict[key] for key in self._attribute_keys] ) def expire_handler( state: InstanceState[Any], keys: Optional[Sequence[str]] ) -> None: if keys is None or set(self._attribute_keys).intersection(keys): state.dict.pop(self.key, None) def insert_update_handler( mapper: Mapper[Any], connection: Connection, state: InstanceState[Any], ) -> None: """After an insert or update, some columns may be expired due to server side defaults, or re-populated due to client side defaults. Pop out the composite value here so that it recreates. """ state.dict.pop(self.key, None) event.listen( self.parent, "after_insert", insert_update_handler, raw=True ) event.listen( self.parent, "after_update", insert_update_handler, raw=True ) event.listen( self.parent, "load", load_handler, raw=True, propagate=True ) event.listen( self.parent, "refresh", refresh_handler, raw=True, propagate=True ) event.listen( self.parent, "expire", expire_handler, raw=True, propagate=True ) proxy_attr = self.parent.class_manager[self.key] proxy_attr.impl.dispatch = proxy_attr.dispatch # type: ignore[assignment] # noqa: E501 proxy_attr.impl.dispatch._active_history = self.active_history # TODO: need a deserialize hook here @util.memoized_property def _attribute_keys(self) -> Sequence[str]: return [prop.key for prop in self.props] def _populate_composite_bulk_save_mappings_fn( self, ) -> Callable[[Dict[str, Any]], None]: if self._generated_composite_accessor: get_values = self._generated_composite_accessor else: def get_values(val: Any) -> Tuple[Any]: return val.__composite_values__() # type: ignore[no-any-return] # noqa: E501 attrs = [prop.key for prop in self.props] def populate(dest_dict: Dict[str, Any]) -> None: dest_dict.update( { key: val for key, val in zip( attrs, get_values(dest_dict.pop(self.key)) ) } ) return populate def get_history( self, state: InstanceState[Any], dict_: _InstanceDict, passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, ) -> History: """Provided for userland code that uses attributes.get_history().""" added: List[Any] = [] deleted: List[Any] = [] has_history = False for prop in self.props: key = prop.key hist = state.manager[key].impl.get_history(state, dict_) if hist.has_changes(): has_history = True non_deleted = hist.non_deleted() if non_deleted: added.extend(non_deleted) else: added.append(None) if hist.deleted: deleted.extend(hist.deleted) else: deleted.append(None) if has_history: return attributes.History( [self._construct_composite(*added)], (), [self._construct_composite(*deleted)], ) else: return attributes.History( (), [self._construct_composite(*added)], () ) def _comparator_factory( self, mapper: Mapper[Any] ) -> Composite.Comparator[_CC]: return self.comparator_factory(self, mapper) class CompositeBundle(orm_util.Bundle[_T]): def __init__( self, property_: Composite[_T], expr: ClauseList, ): self.property = property_ super().__init__(property_.key, *expr) def create_row_processor( self, query: Select[Unpack[TupleAny]], procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]], labels: Sequence[str], ) -> Callable[[Row[Unpack[TupleAny]]], Any]: def proc(row: Row[Unpack[TupleAny]]) -> Any: return self.property._construct_composite( *[proc(row) for proc in procs] ) return proc class Comparator(PropComparator[_PT]): """Produce boolean, comparison, and other operators for :class:`.Composite` attributes. See the example in :ref:`composite_operations` for an overview of usage , as well as the documentation for :class:`.PropComparator`. .. seealso:: :class:`.PropComparator` :class:`.ColumnOperators` :ref:`types_operators` :attr:`.TypeEngine.comparator_factory` """ # https://github.com/python/mypy/issues/4266 __hash__ = None # type: ignore[assignment] prop: RODescriptorReference[Composite[_PT]] @util.memoized_property def clauses(self) -> ClauseList: return expression.ClauseList( group=False, *self._comparable_elements ) def __clause_element__(self) -> CompositeProperty.CompositeBundle[_PT]: return self.expression @util.memoized_property def expression(self) -> CompositeProperty.CompositeBundle[_PT]: clauses = self.clauses._annotate( { "parententity": self._parententity, "parentmapper": self._parententity, "proxy_key": self.prop.key, } ) return CompositeProperty.CompositeBundle(self.prop, clauses) def _bulk_update_tuples( self, value: Any ) -> Sequence[Tuple[_DMLColumnArgument, Any]]: if isinstance(value, BindParameter): value = value.value values: Sequence[Any] if value is None: values = [None for key in self.prop._attribute_keys] elif isinstance(self.prop.composite_class, type) and isinstance( value, self.prop.composite_class ): values = self.prop._composite_values_from_instance( value # type: ignore[arg-type] ) else: raise sa_exc.ArgumentError( "Can't UPDATE composite attribute %s to %r" % (self.prop, value) ) return list(zip(self._comparable_elements, values)) def _bulk_dml_setter(self, key: str) -> Optional[Callable[..., Any]]: return self.prop._populate_composite_bulk_save_mappings_fn() @util.memoized_property def _comparable_elements(self) -> Sequence[QueryableAttribute[Any]]: if self._adapt_to_entity: return [ getattr(self._adapt_to_entity.entity, prop.key) for prop in self.prop._comparable_elements ] else: return self.prop._comparable_elements def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501 return self._compare(operators.eq, other) def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501 return self._compare(operators.ne, other) def __lt__(self, other: Any) -> ColumnElement[bool]: return self._compare(operators.lt, other) def __gt__(self, other: Any) -> ColumnElement[bool]: return self._compare(operators.gt, other) def __le__(self, other: Any) -> ColumnElement[bool]: return self._compare(operators.le, other) def __ge__(self, other: Any) -> ColumnElement[bool]: return self._compare(operators.ge, other) def desc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501 return expression.OrderByList( [e.desc() for e in self._comparable_elements] ) def asc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501 return expression.OrderByList( [e.asc() for e in self._comparable_elements] ) def nulls_first(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501 return expression.OrderByList( [e.nulls_first() for e in self._comparable_elements] ) def nulls_last(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501 return expression.OrderByList( [e.nulls_last() for e in self._comparable_elements] ) # what might be interesting would be if we create # an instance of the composite class itself with # the columns as data members, then use "hybrid style" comparison # to create these comparisons. then your Point.__eq__() method could # be where comparison behavior is defined for SQL also. Likely # not a good choice for default behavior though, not clear how it would # work w/ dataclasses, etc. also no demand for any of this anyway. def _compare( self, operator: OperatorType, other: Any ) -> ColumnElement[bool]: values: Sequence[Any] if other is None: values = [None] * len(self.prop._comparable_elements) else: values = self.prop._composite_values_from_instance(other) comparisons = [ operator(a, b) for a, b in zip(self.prop._comparable_elements, values) ] if self._adapt_to_entity: assert self.adapter is not None comparisons = [self.adapter(x) for x in comparisons] return sql.and_(*comparisons) def __str__(self) -> str: return str(self.parent.class_.__name__) + "." + self.key class Composite(CompositeProperty[_T], _DeclarativeMapped[_T]): """Declarative-compatible front-end for the :class:`.CompositeProperty` class. Public constructor is the :func:`_orm.composite` function. .. versionchanged:: 2.0 Added :class:`_orm.Composite` as a Declarative compatible subclass of :class:`_orm.CompositeProperty`. .. seealso:: :ref:`mapper_composite` """ inherit_cache = True """:meta private:""" class ConcreteInheritedProperty(DescriptorProperty[_T]): """A 'do nothing' :class:`.MapperProperty` that disables an attribute on a concrete subclass that is only present on the inherited mapper, not the concrete classes' mapper. Cases where this occurs include: * When the superclass mapper is mapped against a "polymorphic union", which includes all attributes from all subclasses. * When a relationship() is configured on an inherited mapper, but not on the subclass mapper. Concrete mappers require that relationship() is configured explicitly on each subclass. """ def _comparator_factory( self, mapper: Mapper[Any] ) -> Type[PropComparator[_T]]: comparator_callable = None for m in self.parent.iterate_to_root(): p = m._props[self.key] if getattr(p, "comparator_factory", None) is not None: comparator_callable = p.comparator_factory break assert comparator_callable is not None return comparator_callable(p, mapper) # type: ignore[no-any-return] def __init__(self) -> None: super().__init__() def warn() -> NoReturn: raise AttributeError( "Concrete %s does not implement " "attribute %r at the instance level. Add " "this property explicitly to %s." % (self.parent, self.key, self.parent) ) class NoninheritedConcreteProp: def __set__(s: Any, obj: Any, value: Any) -> NoReturn: warn() def __delete__(s: Any, obj: Any) -> NoReturn: warn() def __get__(s: Any, obj: Any, owner: Any) -> Any: if obj is None: return self.descriptor warn() self.descriptor = NoninheritedConcreteProp() class SynonymProperty(DescriptorProperty[_T]): """Denote an attribute name as a synonym to a mapped property, in that the attribute will mirror the value and expression behavior of another attribute. :class:`.Synonym` is constructed using the :func:`_orm.synonym` function. .. seealso:: :ref:`synonyms` - Overview of synonyms """ comparator_factory: Optional[Type[PropComparator[_T]]] def __init__( self, name: str, map_column: Optional[bool] = None, descriptor: Optional[Any] = None, comparator_factory: Optional[Type[PropComparator[_T]]] = None, attribute_options: Optional[_AttributeOptions] = None, info: Optional[_InfoType] = None, doc: Optional[str] = None, ): super().__init__(attribute_options=attribute_options) self.name = name self.map_column = map_column self.descriptor = descriptor self.comparator_factory = comparator_factory if doc: self.doc = doc elif descriptor and descriptor.__doc__: self.doc = descriptor.__doc__ else: self.doc = None if info: self.info.update(info) util.set_creation_order(self) if not TYPE_CHECKING: @property def uses_objects(self) -> bool: return getattr(self.parent.class_, self.name).impl.uses_objects # TODO: when initialized, check _proxied_object, # emit a warning if its not a column-based property @util.memoized_property def _proxied_object( self, ) -> Union[MapperProperty[_T], SQLORMOperations[_T]]: attr = getattr(self.parent.class_, self.name) if not hasattr(attr, "property") or not isinstance( attr.property, MapperProperty ): # attribute is a non-MapperProprerty proxy such as # hybrid or association proxy if isinstance(attr, attributes.QueryableAttribute): return attr.comparator elif isinstance(attr, SQLORMOperations): # association proxy comes here return attr raise sa_exc.InvalidRequestError( """synonym() attribute "%s.%s" only supports """ """ORM mapped attributes, got %r""" % (self.parent.class_.__name__, self.name, attr) ) return attr.property def _column_strategy_attrs(self) -> Sequence[QueryableAttribute[Any]]: return (getattr(self.parent.class_, self.name),) def _comparator_factory(self, mapper: Mapper[Any]) -> SQLORMOperations[_T]: prop = self._proxied_object if isinstance(prop, MapperProperty): if self.comparator_factory: comp = self.comparator_factory(prop, mapper) else: comp = prop.comparator_factory(prop, mapper) return comp else: return prop def get_history( self, state: InstanceState[Any], dict_: _InstanceDict, passive: PassiveFlag = PassiveFlag.PASSIVE_OFF, ) -> History: attr: QueryableAttribute[Any] = getattr(self.parent.class_, self.name) return attr.impl.get_history(state, dict_, passive=passive) def _get_dataclass_setup_options( self, decl_scan: _ClassScanAbstractConfig, key: str, dataclass_setup_arguments: _DataclassArguments, enable_descriptor_defaults: bool, ) -> _AttributeOptions: dataclasses_default = self._attribute_options.dataclasses_default if ( dataclasses_default is not _NoArg.NO_ARG and not callable(dataclasses_default) and enable_descriptor_defaults and not getattr( decl_scan.cls, "_sa_disable_descriptor_defaults", False ) ): proxied = decl_scan.collected_attributes[self.name] proxied_default = proxied._attribute_options.dataclasses_default if proxied_default != dataclasses_default: raise sa_exc.ArgumentError( f"Synonym {key!r} default argument " f"{dataclasses_default!r} must match the dataclasses " f"default value of proxied object {self.name!r}, " f"""currently { repr(proxied_default) if proxied_default is not _NoArg.NO_ARG else 'not set'}""" ) self._default_scalar_value = dataclasses_default return self._attribute_options._replace( dataclasses_default=DONT_SET ) return self._attribute_options @util.preload_module("sqlalchemy.orm.properties") def set_parent(self, parent: Mapper[Any], init: bool) -> None: properties = util.preloaded.orm_properties if self.map_column: # implement the 'map_column' option. if self.key not in parent.persist_selectable.c: raise sa_exc.ArgumentError( "Can't compile synonym '%s': no column on table " "'%s' named '%s'" % ( self.name, parent.persist_selectable.description, self.key, ) ) elif ( parent.persist_selectable.c[self.key] in parent._columntoproperty and parent._columntoproperty[ parent.persist_selectable.c[self.key] ].key == self.name ): raise sa_exc.ArgumentError( "Can't call map_column=True for synonym %r=%r, " "a ColumnProperty already exists keyed to the name " "%r for column %r" % (self.key, self.name, self.name, self.key) ) p: ColumnProperty[Any] = properties.ColumnProperty( parent.persist_selectable.c[self.key] ) parent._configure_property(self.name, p, init=init, setparent=True) p._mapped_by_synonym = self.key self.parent = parent class Synonym(SynonymProperty[_T], _DeclarativeMapped[_T]): """Declarative front-end for the :class:`.SynonymProperty` class. Public constructor is the :func:`_orm.synonym` function. .. versionchanged:: 2.0 Added :class:`_orm.Synonym` as a Declarative compatible subclass for :class:`_orm.SynonymProperty` .. seealso:: :ref:`synonyms` - Overview of synonyms """ inherit_cache = True """:meta private:"""