diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/orm/properties.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/sqlalchemy/orm/properties.py | 886 |
1 files changed, 886 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/orm/properties.py b/venv/lib/python3.11/site-packages/sqlalchemy/orm/properties.py new file mode 100644 index 0000000..adee44a --- /dev/null +++ b/venv/lib/python3.11/site-packages/sqlalchemy/orm/properties.py @@ -0,0 +1,886 @@ +# orm/properties.py +# Copyright (C) 2005-2024 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 + +"""MapperProperty implementations. + +This is a private module which defines the behavior of individual ORM- +mapped attributes. + +""" + +from __future__ import annotations + +from typing import Any +from typing import cast +from typing import Dict +from typing import List +from typing import Optional +from typing import Sequence +from typing import Set +from typing import Tuple +from typing import Type +from typing import TYPE_CHECKING +from typing import TypeVar +from typing import Union + +from . import attributes +from . import strategy_options +from .base import _DeclarativeMapped +from .base import class_mapper +from .descriptor_props import CompositeProperty +from .descriptor_props import ConcreteInheritedProperty +from .descriptor_props import SynonymProperty +from .interfaces import _AttributeOptions +from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS +from .interfaces import _IntrospectsAnnotations +from .interfaces import _MapsColumns +from .interfaces import MapperProperty +from .interfaces import PropComparator +from .interfaces import StrategizedProperty +from .relationships import RelationshipProperty +from .util import de_stringify_annotation +from .util import de_stringify_union_elements +from .. import exc as sa_exc +from .. import ForeignKey +from .. import log +from .. import util +from ..sql import coercions +from ..sql import roles +from ..sql.base import _NoArg +from ..sql.schema import Column +from ..sql.schema import SchemaConst +from ..sql.type_api import TypeEngine +from ..util.typing import de_optionalize_union_types +from ..util.typing import is_fwd_ref +from ..util.typing import is_optional_union +from ..util.typing import is_pep593 +from ..util.typing import is_pep695 +from ..util.typing import is_union +from ..util.typing import Self +from ..util.typing import typing_get_args + +if TYPE_CHECKING: + from ._typing import _IdentityKeyType + from ._typing import _InstanceDict + from ._typing import _ORMColumnExprArgument + from ._typing import _RegistryType + from .base import Mapped + from .decl_base import _ClassScanMapperConfig + from .mapper import Mapper + from .session import Session + from .state import _InstallLoaderCallableProto + from .state import InstanceState + from ..sql._typing import _InfoType + from ..sql.elements import ColumnElement + from ..sql.elements import NamedColumn + from ..sql.operators import OperatorType + from ..util.typing import _AnnotationScanType + from ..util.typing import RODescriptorReference + +_T = TypeVar("_T", bound=Any) +_PT = TypeVar("_PT", bound=Any) +_NC = TypeVar("_NC", bound="NamedColumn[Any]") + +__all__ = [ + "ColumnProperty", + "CompositeProperty", + "ConcreteInheritedProperty", + "RelationshipProperty", + "SynonymProperty", +] + + +@log.class_logger +class ColumnProperty( + _MapsColumns[_T], + StrategizedProperty[_T], + _IntrospectsAnnotations, + log.Identified, +): + """Describes an object attribute that corresponds to a table column + or other column expression. + + Public constructor is the :func:`_orm.column_property` function. + + """ + + strategy_wildcard_key = strategy_options._COLUMN_TOKEN + inherit_cache = True + """:meta private:""" + + _links_to_entity = False + + columns: List[NamedColumn[Any]] + + _is_polymorphic_discriminator: bool + + _mapped_by_synonym: Optional[str] + + comparator_factory: Type[PropComparator[_T]] + + __slots__ = ( + "columns", + "group", + "deferred", + "instrument", + "comparator_factory", + "active_history", + "expire_on_flush", + "_creation_order", + "_is_polymorphic_discriminator", + "_mapped_by_synonym", + "_deferred_column_loader", + "_raise_column_loader", + "_renders_in_subqueries", + "raiseload", + ) + + def __init__( + self, + column: _ORMColumnExprArgument[_T], + *additional_columns: _ORMColumnExprArgument[Any], + attribute_options: Optional[_AttributeOptions] = None, + group: Optional[str] = None, + deferred: bool = False, + raiseload: bool = False, + comparator_factory: Optional[Type[PropComparator[_T]]] = None, + active_history: bool = False, + expire_on_flush: bool = True, + info: Optional[_InfoType] = None, + doc: Optional[str] = None, + _instrument: bool = True, + _assume_readonly_dc_attributes: bool = False, + ): + super().__init__( + attribute_options=attribute_options, + _assume_readonly_dc_attributes=_assume_readonly_dc_attributes, + ) + columns = (column,) + additional_columns + self.columns = [ + coercions.expect(roles.LabeledColumnExprRole, c) for c in columns + ] + self.group = group + self.deferred = deferred + self.raiseload = raiseload + self.instrument = _instrument + self.comparator_factory = ( + comparator_factory + if comparator_factory is not None + else self.__class__.Comparator + ) + self.active_history = active_history + self.expire_on_flush = expire_on_flush + + if info is not None: + self.info.update(info) + + if doc is not None: + self.doc = doc + else: + for col in reversed(self.columns): + doc = getattr(col, "doc", None) + if doc is not None: + self.doc = doc + break + else: + self.doc = None + + util.set_creation_order(self) + + self.strategy_key = ( + ("deferred", self.deferred), + ("instrument", self.instrument), + ) + if self.raiseload: + self.strategy_key += (("raiseload", True),) + + 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: + column = self.columns[0] + if column.key is None: + column.key = key + if column.name is None: + column.name = key + + @property + def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]: + return self + + @property + def columns_to_assign(self) -> List[Tuple[Column[Any], int]]: + # mypy doesn't care about the isinstance here + return [ + (c, 0) # type: ignore + for c in self.columns + if isinstance(c, Column) and c.table is None + ] + + def _memoized_attr__renders_in_subqueries(self) -> bool: + if ("query_expression", True) in self.strategy_key: + return self.strategy._have_default_expression # type: ignore + + return ("deferred", True) not in self.strategy_key or ( + self not in self.parent._readonly_props # type: ignore + ) + + @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies") + def _memoized_attr__deferred_column_loader( + self, + ) -> _InstallLoaderCallableProto[Any]: + state = util.preloaded.orm_state + strategies = util.preloaded.orm_strategies + return state.InstanceState._instance_level_callable_processor( + self.parent.class_manager, + strategies.LoadDeferredColumns(self.key), + self.key, + ) + + @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies") + def _memoized_attr__raise_column_loader( + self, + ) -> _InstallLoaderCallableProto[Any]: + state = util.preloaded.orm_state + strategies = util.preloaded.orm_strategies + return state.InstanceState._instance_level_callable_processor( + self.parent.class_manager, + strategies.LoadDeferredColumns(self.key, True), + self.key, + ) + + def __clause_element__(self) -> roles.ColumnsClauseRole: + """Allow the ColumnProperty to work in expression before it is turned + into an instrumented attribute. + """ + + return self.expression + + @property + def expression(self) -> roles.ColumnsClauseRole: + """Return the primary column or expression for this ColumnProperty. + + E.g.:: + + + class File(Base): + # ... + + name = Column(String(64)) + extension = Column(String(8)) + filename = column_property(name + '.' + extension) + path = column_property('C:/' + filename.expression) + + .. seealso:: + + :ref:`mapper_column_property_sql_expressions_composed` + + """ + return self.columns[0] + + def instrument_class(self, mapper: Mapper[Any]) -> None: + if not self.instrument: + return + + attributes.register_descriptor( + mapper.class_, + self.key, + comparator=self.comparator_factory(self, mapper), + parententity=mapper, + doc=self.doc, + ) + + def do_init(self) -> None: + super().do_init() + + if len(self.columns) > 1 and set(self.parent.primary_key).issuperset( + self.columns + ): + util.warn( + ( + "On mapper %s, primary key column '%s' is being combined " + "with distinct primary key column '%s' in attribute '%s'. " + "Use explicit properties to give each column its own " + "mapped attribute name." + ) + % (self.parent, self.columns[1], self.columns[0], self.key) + ) + + def copy(self) -> ColumnProperty[_T]: + return ColumnProperty( + *self.columns, + deferred=self.deferred, + group=self.group, + active_history=self.active_history, + ) + + def merge( + self, + session: Session, + source_state: InstanceState[Any], + source_dict: _InstanceDict, + dest_state: InstanceState[Any], + dest_dict: _InstanceDict, + load: bool, + _recursive: Dict[Any, object], + _resolve_conflict_map: Dict[_IdentityKeyType[Any], object], + ) -> None: + if not self.instrument: + return + elif self.key in source_dict: + value = source_dict[self.key] + + if not load: + dest_dict[self.key] = value + else: + impl = dest_state.get_impl(self.key) + impl.set(dest_state, dest_dict, value, None) + elif dest_state.has_identity and self.key not in dest_dict: + dest_state._expire_attributes( + dest_dict, [self.key], no_loader=True + ) + + class Comparator(util.MemoizedSlots, PropComparator[_PT]): + """Produce boolean, comparison, and other operators for + :class:`.ColumnProperty` attributes. + + See the documentation for :class:`.PropComparator` for a brief + overview. + + .. seealso:: + + :class:`.PropComparator` + + :class:`.ColumnOperators` + + :ref:`types_operators` + + :attr:`.TypeEngine.comparator_factory` + + """ + + if not TYPE_CHECKING: + # prevent pylance from being clever about slots + __slots__ = "__clause_element__", "info", "expressions" + + prop: RODescriptorReference[ColumnProperty[_PT]] + + expressions: Sequence[NamedColumn[Any]] + """The full sequence of columns referenced by this + attribute, adjusted for any aliasing in progress. + + .. versionadded:: 1.3.17 + + .. seealso:: + + :ref:`maptojoin` - usage example + """ + + def _orm_annotate_column(self, column: _NC) -> _NC: + """annotate and possibly adapt a column to be returned + as the mapped-attribute exposed version of the column. + + The column in this context needs to act as much like the + column in an ORM mapped context as possible, so includes + annotations to give hints to various ORM functions as to + the source entity of this column. It also adapts it + to the mapper's with_polymorphic selectable if one is + present. + + """ + + pe = self._parententity + annotations: Dict[str, Any] = { + "entity_namespace": pe, + "parententity": pe, + "parentmapper": pe, + "proxy_key": self.prop.key, + } + + col = column + + # for a mapper with polymorphic_on and an adapter, return + # the column against the polymorphic selectable. + # see also orm.util._orm_downgrade_polymorphic_columns + # for the reverse operation. + if self._parentmapper._polymorphic_adapter: + mapper_local_col = col + col = self._parentmapper._polymorphic_adapter.traverse(col) + + # this is a clue to the ORM Query etc. that this column + # was adapted to the mapper's polymorphic_adapter. the + # ORM uses this hint to know which column its adapting. + annotations["adapt_column"] = mapper_local_col + + return col._annotate(annotations)._set_propagate_attrs( + {"compile_state_plugin": "orm", "plugin_subject": pe} + ) + + if TYPE_CHECKING: + + def __clause_element__(self) -> NamedColumn[_PT]: ... + + def _memoized_method___clause_element__( + self, + ) -> NamedColumn[_PT]: + if self.adapter: + return self.adapter(self.prop.columns[0], self.prop.key) + else: + return self._orm_annotate_column(self.prop.columns[0]) + + def _memoized_attr_info(self) -> _InfoType: + """The .info dictionary for this attribute.""" + + ce = self.__clause_element__() + try: + return ce.info # type: ignore + except AttributeError: + return self.prop.info + + def _memoized_attr_expressions(self) -> Sequence[NamedColumn[Any]]: + """The full sequence of columns referenced by this + attribute, adjusted for any aliasing in progress. + + .. versionadded:: 1.3.17 + + """ + if self.adapter: + return [ + self.adapter(col, self.prop.key) + for col in self.prop.columns + ] + else: + return [ + self._orm_annotate_column(col) for col in self.prop.columns + ] + + def _fallback_getattr(self, key: str) -> Any: + """proxy attribute access down to the mapped column. + + this allows user-defined comparison methods to be accessed. + """ + return getattr(self.__clause_element__(), key) + + def operate( + self, op: OperatorType, *other: Any, **kwargs: Any + ) -> ColumnElement[Any]: + return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501 + + def reverse_operate( + self, op: OperatorType, other: Any, **kwargs: Any + ) -> ColumnElement[Any]: + col = self.__clause_element__() + return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501 + + def __str__(self) -> str: + if not self.parent or not self.key: + return object.__repr__(self) + return str(self.parent.class_.__name__) + "." + self.key + + +class MappedSQLExpression(ColumnProperty[_T], _DeclarativeMapped[_T]): + """Declarative front-end for the :class:`.ColumnProperty` class. + + Public constructor is the :func:`_orm.column_property` function. + + .. versionchanged:: 2.0 Added :class:`_orm.MappedSQLExpression` as + a Declarative compatible subclass for :class:`_orm.ColumnProperty`. + + .. seealso:: + + :class:`.MappedColumn` + + """ + + inherit_cache = True + """:meta private:""" + + +class MappedColumn( + _IntrospectsAnnotations, + _MapsColumns[_T], + _DeclarativeMapped[_T], +): + """Maps a single :class:`_schema.Column` on a class. + + :class:`_orm.MappedColumn` is a specialization of the + :class:`_orm.ColumnProperty` class and is oriented towards declarative + configuration. + + To construct :class:`_orm.MappedColumn` objects, use the + :func:`_orm.mapped_column` constructor function. + + .. versionadded:: 2.0 + + + """ + + __slots__ = ( + "column", + "_creation_order", + "_sort_order", + "foreign_keys", + "_has_nullable", + "_has_insert_default", + "deferred", + "deferred_group", + "deferred_raiseload", + "active_history", + "_attribute_options", + "_has_dataclass_arguments", + "_use_existing_column", + ) + + deferred: Union[_NoArg, bool] + deferred_raiseload: bool + deferred_group: Optional[str] + + column: Column[_T] + foreign_keys: Optional[Set[ForeignKey]] + _attribute_options: _AttributeOptions + + def __init__(self, *arg: Any, **kw: Any): + self._attribute_options = attr_opts = kw.pop( + "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS + ) + + self._use_existing_column = kw.pop("use_existing_column", False) + + self._has_dataclass_arguments = ( + attr_opts is not None + and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS + and any( + attr_opts[i] is not _NoArg.NO_ARG + for i, attr in enumerate(attr_opts._fields) + if attr != "dataclasses_default" + ) + ) + + insert_default = kw.pop("insert_default", _NoArg.NO_ARG) + self._has_insert_default = insert_default is not _NoArg.NO_ARG + + if self._has_insert_default: + kw["default"] = insert_default + elif attr_opts.dataclasses_default is not _NoArg.NO_ARG: + kw["default"] = attr_opts.dataclasses_default + + self.deferred_group = kw.pop("deferred_group", None) + self.deferred_raiseload = kw.pop("deferred_raiseload", None) + self.deferred = kw.pop("deferred", _NoArg.NO_ARG) + self.active_history = kw.pop("active_history", False) + + self._sort_order = kw.pop("sort_order", _NoArg.NO_ARG) + self.column = cast("Column[_T]", Column(*arg, **kw)) + self.foreign_keys = self.column.foreign_keys + self._has_nullable = "nullable" in kw and kw.get("nullable") not in ( + None, + SchemaConst.NULL_UNSPECIFIED, + ) + util.set_creation_order(self) + + def _copy(self, **kw: Any) -> Self: + new = self.__class__.__new__(self.__class__) + new.column = self.column._copy(**kw) + new.deferred = self.deferred + new.deferred_group = self.deferred_group + new.deferred_raiseload = self.deferred_raiseload + new.foreign_keys = new.column.foreign_keys + new.active_history = self.active_history + new._has_nullable = self._has_nullable + new._attribute_options = self._attribute_options + new._has_insert_default = self._has_insert_default + new._has_dataclass_arguments = self._has_dataclass_arguments + new._use_existing_column = self._use_existing_column + new._sort_order = self._sort_order + util.set_creation_order(new) + return new + + @property + def name(self) -> str: + return self.column.name + + @property + def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]: + effective_deferred = self.deferred + if effective_deferred is _NoArg.NO_ARG: + effective_deferred = bool( + self.deferred_group or self.deferred_raiseload + ) + + if effective_deferred or self.active_history: + return ColumnProperty( + self.column, + deferred=effective_deferred, + group=self.deferred_group, + raiseload=self.deferred_raiseload, + attribute_options=self._attribute_options, + active_history=self.active_history, + ) + else: + return None + + @property + def columns_to_assign(self) -> List[Tuple[Column[Any], int]]: + return [ + ( + self.column, + ( + self._sort_order + if self._sort_order is not _NoArg.NO_ARG + else 0 + ), + ) + ] + + def __clause_element__(self) -> Column[_T]: + return self.column + + def operate( + self, op: OperatorType, *other: Any, **kwargs: Any + ) -> ColumnElement[Any]: + return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501 + + def reverse_operate( + self, op: OperatorType, other: Any, **kwargs: Any + ) -> ColumnElement[Any]: + col = self.__clause_element__() + return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501 + + def found_in_pep593_annotated(self) -> Any: + # return a blank mapped_column(). This mapped_column()'s + # Column will be merged into it in _init_column_for_annotation(). + return MappedColumn() + + 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: + column = self.column + + if ( + self._use_existing_column + and decl_scan.inherits + and decl_scan.single + ): + if decl_scan.is_deferred: + raise sa_exc.ArgumentError( + "Can't use use_existing_column with deferred mappers" + ) + supercls_mapper = class_mapper(decl_scan.inherits, False) + + colname = column.name if column.name is not None else key + column = self.column = supercls_mapper.local_table.c.get( # type: ignore # noqa: E501 + colname, column + ) + + if column.key is None: + column.key = key + if column.name is None: + column.name = key + + sqltype = column.type + + if extracted_mapped_annotation is None: + if sqltype._isnull and not self.column.foreign_keys: + self._raise_for_required(key, cls) + else: + return + + self._init_column_for_annotation( + cls, + registry, + extracted_mapped_annotation, + originating_module, + ) + + @util.preload_module("sqlalchemy.orm.decl_base") + def declarative_scan_for_composite( + self, + registry: _RegistryType, + cls: Type[Any], + originating_module: Optional[str], + key: str, + param_name: str, + param_annotation: _AnnotationScanType, + ) -> None: + decl_base = util.preloaded.orm_decl_base + decl_base._undefer_column_name(param_name, self.column) + self._init_column_for_annotation( + cls, registry, param_annotation, originating_module + ) + + def _init_column_for_annotation( + self, + cls: Type[Any], + registry: _RegistryType, + argument: _AnnotationScanType, + originating_module: Optional[str], + ) -> None: + sqltype = self.column.type + + if isinstance(argument, str) or is_fwd_ref( + argument, check_generic=True + ): + assert originating_module is not None + argument = de_stringify_annotation( + cls, argument, originating_module, include_generic=True + ) + + if is_union(argument): + assert originating_module is not None + argument = de_stringify_union_elements( + cls, argument, originating_module + ) + + nullable = is_optional_union(argument) + + if not self._has_nullable: + self.column.nullable = nullable + + our_type = de_optionalize_union_types(argument) + + use_args_from = None + + our_original_type = our_type + + if is_pep695(our_type): + our_type = our_type.__value__ + + if is_pep593(our_type): + our_type_is_pep593 = True + + pep_593_components = typing_get_args(our_type) + raw_pep_593_type = pep_593_components[0] + if is_optional_union(raw_pep_593_type): + raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) + + nullable = True + if not self._has_nullable: + self.column.nullable = nullable + for elem in pep_593_components[1:]: + if isinstance(elem, MappedColumn): + use_args_from = elem + break + else: + our_type_is_pep593 = False + raw_pep_593_type = None + + if use_args_from is not None: + if ( + not self._has_insert_default + and use_args_from.column.default is not None + ): + self.column.default = None + + use_args_from.column._merge(self.column) + sqltype = self.column.type + + if ( + use_args_from.deferred is not _NoArg.NO_ARG + and self.deferred is _NoArg.NO_ARG + ): + self.deferred = use_args_from.deferred + + if ( + use_args_from.deferred_group is not None + and self.deferred_group is None + ): + self.deferred_group = use_args_from.deferred_group + + if ( + use_args_from.deferred_raiseload is not None + and self.deferred_raiseload is None + ): + self.deferred_raiseload = use_args_from.deferred_raiseload + + if ( + use_args_from._use_existing_column + and not self._use_existing_column + ): + self._use_existing_column = True + + if use_args_from.active_history: + self.active_history = use_args_from.active_history + + if ( + use_args_from._sort_order is not None + and self._sort_order is _NoArg.NO_ARG + ): + self._sort_order = use_args_from._sort_order + + if ( + use_args_from.column.key is not None + or use_args_from.column.name is not None + ): + util.warn_deprecated( + "Can't use the 'key' or 'name' arguments in " + "Annotated with mapped_column(); this will be ignored", + "2.0.22", + ) + + if use_args_from._has_dataclass_arguments: + for idx, arg in enumerate( + use_args_from._attribute_options._fields + ): + if ( + use_args_from._attribute_options[idx] + is not _NoArg.NO_ARG + ): + arg = arg.replace("dataclasses_", "") + util.warn_deprecated( + f"Argument '{arg}' is a dataclass argument and " + "cannot be specified within a mapped_column() " + "bundled inside of an Annotated object", + "2.0.22", + ) + + if sqltype._isnull and not self.column.foreign_keys: + new_sqltype = None + + if our_type_is_pep593: + checks = [our_original_type, raw_pep_593_type] + else: + checks = [our_original_type] + + for check_type in checks: + new_sqltype = registry._resolve_type(check_type) + if new_sqltype is not None: + break + else: + if isinstance(our_type, TypeEngine) or ( + isinstance(our_type, type) + and issubclass(our_type, TypeEngine) + ): + raise sa_exc.ArgumentError( + f"The type provided inside the {self.column.key!r} " + "attribute Mapped annotation is the SQLAlchemy type " + f"{our_type}. Expected a Python type instead" + ) + else: + raise sa_exc.ArgumentError( + "Could not locate SQLAlchemy Core type for Python " + f"type {our_type} inside the {self.column.key!r} " + "attribute Mapped annotation" + ) + + self.column._set_type(new_sqltype) |