diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/sqlalchemy/orm/descriptor_props.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/sqlalchemy/orm/descriptor_props.py | 1077 |
1 files changed, 1077 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sqlalchemy/orm/descriptor_props.py b/.venv/lib/python3.12/site-packages/sqlalchemy/orm/descriptor_props.py new file mode 100644 index 00000000..f01cc178 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/sqlalchemy/orm/descriptor_props.py @@ -0,0 +1,1077 @@ +# orm/descriptor_props.py +# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors +# <see AUTHORS file> +# +# 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 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 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 _none_set +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.elements import BindParameter +from ..util.typing import get_args +from ..util.typing import is_fwd_ref +from ..util.typing import is_pep593 + + +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 _ClassScanMapperConfig + 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 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 + 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], ...] + + _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], + attribute_options: Optional[_AttributeOptions] = None, + active_history: bool = False, + deferred: bool = False, + group: Optional[str] = None, + comparator_factory: Optional[Type[Comparator[_CC]]] = 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 + else: + self.composite_class = _class_or_attr # type: ignore + self.attrs = attrs + + 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() + + 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 + + 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 + ] + + # current expected behavior here is that the composite is + # created on access if the object is persistent or if + # col attributes have non-None. This would be better + # if the composite were created unconditionally, + # but that would be a behavioral change. + if self.key not in dict_ and ( + state.key is not None or not _none_set.issuperset(values) + ): + dict_[self.key] = self.composite_class(*values) + state.manager.dispatch.refresh( + state, self._COMPOSITE_FGET, [self.key] + ) + + return dict_.get(self.key, None) + + def fset(instance: Any, value: Any) -> None: + 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: _ClassScanMapperConfig, + 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 + ) + + self.composite_class = argument + + if is_dataclass(self.composite_class): + self._setup_for_dataclass(registry, cls, originating_module, key) + else: + 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, + 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 + attr = MappedColumn() + self.attrs += (attr,) + + if isinstance(attr, MappedColumn): + attr.declarative_scan_for_composite( + 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 + + @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.composite_class( + *[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 + 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 + + 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.composite_class(*added)], + (), + [self.composite_class(*deleted)], + ) + else: + return attributes.History((), [self.composite_class(*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[Any], + procs: Sequence[Callable[[Row[Any]], Any]], + labels: Sequence[str], + ) -> Callable[[Row[Any]], Any]: + def proc(row: Row[Any]) -> Any: + return self.property.composite_class( + *[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 + + 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)) + + @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) + + # 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 + + 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): + # assocaition 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 _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) + + @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:""" |