codekingpro/portable-devtools
114k
1# orm/properties.py
2# Copyright (C) 2005-2026 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"""MapperProperty implementations.
9
10This is a private module which defines the behavior of individual ORM-
11mapped attributes.
12
13"""
14
15from __future__ import annotations
16
17from typing import Any
18from typing import cast
19from typing import Dict
20from typing import List
21from typing import Optional
22from typing import Sequence
23from typing import Set
24from typing import Tuple
25from typing import Type
26from typing import TYPE_CHECKING
27from typing import TypeVar
28from typing import Union
29
30from . import attributes
31from . import exc as orm_exc
32from . import strategy_options
33from .base import _DeclarativeMapped
34from .base import class_mapper
35from .descriptor_props import CompositeProperty
36from .descriptor_props import ConcreteInheritedProperty
37from .descriptor_props import SynonymProperty
38from .interfaces import _AttributeOptions
39from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS
40from .interfaces import _IntrospectsAnnotations
41from .interfaces import _MapsColumns
42from .interfaces import MapperProperty
43from .interfaces import PropComparator
44from .interfaces import StrategizedProperty
45from .relationships import RelationshipProperty
46from .util import de_stringify_annotation
47from .. import exc as sa_exc
48from .. import ForeignKey
49from .. import log
50from .. import util
51from ..sql import coercions
52from ..sql import roles
53from ..sql.base import _NoArg
54from ..sql.schema import Column
55from ..sql.schema import SchemaConst
56from ..sql.type_api import TypeEngine
57from ..util.typing import de_optionalize_union_types
58from ..util.typing import get_args
59from ..util.typing import includes_none
60from ..util.typing import is_a_type
61from ..util.typing import is_fwd_ref
62from ..util.typing import is_pep593
63from ..util.typing import is_pep695
64from ..util.typing import Self
65
66if TYPE_CHECKING:
67 from ._typing import _IdentityKeyType
68 from ._typing import _InstanceDict
69 from ._typing import _ORMColumnExprArgument
70 from ._typing import _RegistryType
71 from .base import Mapped
72 from .decl_base import _ClassScanMapperConfig
73 from .mapper import Mapper
74 from .session import Session
75 from .state import _InstallLoaderCallableProto
76 from .state import InstanceState
77 from ..sql._typing import _InfoType
78 from ..sql.elements import ColumnElement
79 from ..sql.elements import NamedColumn
80 from ..sql.operators import OperatorType
81 from ..util.typing import _AnnotationScanType
82 from ..util.typing import RODescriptorReference
83
84_T = TypeVar("_T", bound=Any)
85_PT = TypeVar("_PT", bound=Any)
86_NC = TypeVar("_NC", bound="NamedColumn[Any]")
87
88__all__ = [
89 "ColumnProperty",
90 "CompositeProperty",
91 "ConcreteInheritedProperty",
92 "RelationshipProperty",
93 "SynonymProperty",
94]
95
96
97@log.class_logger
98class ColumnProperty(
99 _MapsColumns[_T],
100 StrategizedProperty[_T],
101 _IntrospectsAnnotations,
102 log.Identified,
103):
104 """Describes an object attribute that corresponds to a table column
105 or other column expression.
106
107 Public constructor is the :func:`_orm.column_property` function.
108
109 """
110
111 strategy_wildcard_key = strategy_options._COLUMN_TOKEN
112 inherit_cache = True
113 """:meta private:"""
114
115 _links_to_entity = False
116
117 columns: List[NamedColumn[Any]]
118
119 _is_polymorphic_discriminator: bool
120
121 _mapped_by_synonym: Optional[str]
122
123 comparator_factory: Type[PropComparator[_T]]
124
125 __slots__ = (
126 "columns",
127 "group",
128 "deferred",
129 "instrument",
130 "comparator_factory",
131 "active_history",
132 "expire_on_flush",
133 "_creation_order",
134 "_is_polymorphic_discriminator",
135 "_mapped_by_synonym",
136 "_deferred_column_loader",
137 "_raise_column_loader",
138 "_renders_in_subqueries",
139 "raiseload",
140 )
141
142 def __init__(
143 self,
144 column: _ORMColumnExprArgument[_T],
145 *additional_columns: _ORMColumnExprArgument[Any],
146 attribute_options: Optional[_AttributeOptions] = None,
147 group: Optional[str] = None,
148 deferred: bool = False,
149 raiseload: bool = False,
150 comparator_factory: Optional[Type[PropComparator[_T]]] = None,
151 active_history: bool = False,
152 expire_on_flush: bool = True,
153 info: Optional[_InfoType] = None,
154 doc: Optional[str] = None,
155 _instrument: bool = True,
156 _assume_readonly_dc_attributes: bool = False,
157 ):
158 super().__init__(
159 attribute_options=attribute_options,
160 _assume_readonly_dc_attributes=_assume_readonly_dc_attributes,
161 )
162 columns = (column,) + additional_columns
163 self.columns = [
164 coercions.expect(roles.LabeledColumnExprRole, c) for c in columns
165 ]
166 self.group = group
167 self.deferred = deferred
168 self.raiseload = raiseload
169 self.instrument = _instrument
170 self.comparator_factory = (
171 comparator_factory
172 if comparator_factory is not None
173 else self.__class__.Comparator
174 )
175 self.active_history = active_history
176 self.expire_on_flush = expire_on_flush
177
178 if info is not None:
179 self.info.update(info)
180
181 if doc is not None:
182 self.doc = doc
183 else:
184 for col in reversed(self.columns):
185 doc = getattr(col, "doc", None)
186 if doc is not None:
187 self.doc = doc
188 break
189 else:
190 self.doc = None
191
192 util.set_creation_order(self)
193
194 self.strategy_key = (
195 ("deferred", self.deferred),
196 ("instrument", self.instrument),
197 )
198 if self.raiseload:
199 self.strategy_key += (("raiseload", True),)
200
201 def declarative_scan(
202 self,
203 decl_scan: _ClassScanMapperConfig,
204 registry: _RegistryType,
205 cls: Type[Any],
206 originating_module: Optional[str],
207 key: str,
208 mapped_container: Optional[Type[Mapped[Any]]],
209 annotation: Optional[_AnnotationScanType],
210 extracted_mapped_annotation: Optional[_AnnotationScanType],
211 is_dataclass_field: bool,
212 ) -> None:
213 column = self.columns[0]
214 if column.key is None:
215 column.key = key
216 if column.name is None:
217 column.name = key
218
219 @property
220 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
221 return self
222
223 @property
224 def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
225 # mypy doesn't care about the isinstance here
226 return [
227 (c, 0) # type: ignore
228 for c in self.columns
229 if isinstance(c, Column) and c.table is None
230 ]
231
232 def _memoized_attr__renders_in_subqueries(self) -> bool:
233 if ("query_expression", True) in self.strategy_key:
234 return self.strategy._have_default_expression # type: ignore
235
236 return ("deferred", True) not in self.strategy_key or (
237 self not in self.parent._readonly_props
238 )
239
240 @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies")
241 def _memoized_attr__deferred_column_loader(
242 self,
243 ) -> _InstallLoaderCallableProto[Any]:
244 state = util.preloaded.orm_state
245 strategies = util.preloaded.orm_strategies
246 return state.InstanceState._instance_level_callable_processor(
247 self.parent.class_manager,
248 strategies.LoadDeferredColumns(self.key),
249 self.key,
250 )
251
252 @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies")
253 def _memoized_attr__raise_column_loader(
254 self,
255 ) -> _InstallLoaderCallableProto[Any]:
256 state = util.preloaded.orm_state
257 strategies = util.preloaded.orm_strategies
258 return state.InstanceState._instance_level_callable_processor(
259 self.parent.class_manager,
260 strategies.LoadDeferredColumns(self.key, True),
261 self.key,
262 )
263
264 def __clause_element__(self) -> roles.ColumnsClauseRole:
265 """Allow the ColumnProperty to work in expression before it is turned
266 into an instrumented attribute.
267 """
268
269 return self.expression
270
271 @property
272 def expression(self) -> roles.ColumnsClauseRole:
273 """Return the primary column or expression for this ColumnProperty.
274
275 E.g.::
276
277
278 class File(Base):
279 # ...
280
281 name = Column(String(64))
282 extension = Column(String(8))
283 filename = column_property(name + "." + extension)
284 path = column_property("C:/" + filename.expression)
285
286 .. seealso::
287
288 :ref:`mapper_column_property_sql_expressions_composed`
289
290 """
291 return self.columns[0]
292
293 def instrument_class(self, mapper: Mapper[Any]) -> None:
294 if not self.instrument:
295 return
296
297 attributes.register_descriptor(
298 mapper.class_,
299 self.key,
300 comparator=self.comparator_factory(self, mapper),
301 parententity=mapper,
302 doc=self.doc,
303 )
304
305 def do_init(self) -> None:
306 super().do_init()
307
308 if len(self.columns) > 1 and set(self.parent.primary_key).issuperset(
309 self.columns
310 ):
311 util.warn(
312 (
313 "On mapper %s, primary key column '%s' is being combined "
314 "with distinct primary key column '%s' in attribute '%s'. "
315 "Use explicit properties to give each column its own "
316 "mapped attribute name."
317 )
318 % (self.parent, self.columns[1], self.columns[0], self.key)
319 )
320
321 def copy(self) -> ColumnProperty[_T]:
322 return ColumnProperty(
323 *self.columns,
324 deferred=self.deferred,
325 group=self.group,
326 active_history=self.active_history,
327 )
328
329 def merge(
330 self,
331 session: Session,
332 source_state: InstanceState[Any],
333 source_dict: _InstanceDict,
334 dest_state: InstanceState[Any],
335 dest_dict: _InstanceDict,
336 load: bool,
337 _recursive: Dict[Any, object],
338 _resolve_conflict_map: Dict[_IdentityKeyType[Any], object],
339 ) -> None:
340 if not self.instrument:
341 return
342 elif self.key in source_dict:
343 value = source_dict[self.key]
344
345 if not load:
346 dest_dict[self.key] = value
347 else:
348 impl = dest_state.get_impl(self.key)
349 impl.set(dest_state, dest_dict, value, None)
350 elif dest_state.has_identity and self.key not in dest_dict:
351 dest_state._expire_attributes(
352 dest_dict, [self.key], no_loader=True
353 )
354
355 class Comparator(util.MemoizedSlots, PropComparator[_PT]):
356 """Produce boolean, comparison, and other operators for
357 :class:`.ColumnProperty` attributes.
358
359 See the documentation for :class:`.PropComparator` for a brief
360 overview.
361
362 .. seealso::
363
364 :class:`.PropComparator`
365
366 :class:`.ColumnOperators`
367
368 :ref:`types_operators`
369
370 :attr:`.TypeEngine.comparator_factory`
371
372 """
373
374 if not TYPE_CHECKING:
375 # prevent pylance from being clever about slots
376 __slots__ = "__clause_element__", "info", "expressions"
377
378 prop: RODescriptorReference[ColumnProperty[_PT]]
379
380 expressions: Sequence[NamedColumn[Any]]
381 """The full sequence of columns referenced by this
382 attribute, adjusted for any aliasing in progress.
383
384 .. versionadded:: 1.3.17
385
386 .. seealso::
387
388 :ref:`maptojoin` - usage example
389 """
390
391 def _orm_annotate_column(self, column: _NC) -> _NC:
392 """annotate and possibly adapt a column to be returned
393 as the mapped-attribute exposed version of the column.
394
395 The column in this context needs to act as much like the
396 column in an ORM mapped context as possible, so includes
397 annotations to give hints to various ORM functions as to
398 the source entity of this column. It also adapts it
399 to the mapper's with_polymorphic selectable if one is
400 present.
401
402 """
403
404 pe = self._parententity
405 annotations: Dict[str, Any] = {
406 "entity_namespace": pe,
407 "parententity": pe,
408 "parentmapper": pe,
409 "proxy_key": self.prop.key,
410 }
411
412 col = column
413
414 # for a mapper with polymorphic_on and an adapter, return
415 # the column against the polymorphic selectable.
416 # see also orm.util._orm_downgrade_polymorphic_columns
417 # for the reverse operation.
418 if self._parentmapper._polymorphic_adapter:
419 mapper_local_col = col
420 col = self._parentmapper._polymorphic_adapter.traverse(col)
421
422 # this is a clue to the ORM Query etc. that this column
423 # was adapted to the mapper's polymorphic_adapter. the
424 # ORM uses this hint to know which column its adapting.
425 annotations["adapt_column"] = mapper_local_col
426
427 return col._annotate(annotations)._set_propagate_attrs(
428 {"compile_state_plugin": "orm", "plugin_subject": pe}
429 )
430
431 if TYPE_CHECKING:
432
433 def __clause_element__(self) -> NamedColumn[_PT]: ...
434
435 def _memoized_method___clause_element__(
436 self,
437 ) -> NamedColumn[_PT]:
438 if self.adapter:
439 return self.adapter(self.prop.columns[0], self.prop.key)
440 else:
441 return self._orm_annotate_column(self.prop.columns[0])
442
443 def _memoized_attr_info(self) -> _InfoType:
444 """The .info dictionary for this attribute."""
445
446 ce = self.__clause_element__()
447 try:
448 return ce.info # type: ignore
449 except AttributeError:
450 return self.prop.info
451
452 def _memoized_attr_expressions(self) -> Sequence[NamedColumn[Any]]:
453 """The full sequence of columns referenced by this
454 attribute, adjusted for any aliasing in progress.
455
456 .. versionadded:: 1.3.17
457
458 """
459 if self.adapter:
460 return [
461 self.adapter(col, self.prop.key)
462 for col in self.prop.columns
463 ]
464 else:
465 return [
466 self._orm_annotate_column(col) for col in self.prop.columns
467 ]
468
469 def _fallback_getattr(self, key: str) -> Any:
470 """proxy attribute access down to the mapped column.
471
472 this allows user-defined comparison methods to be accessed.
473 """
474 return getattr(self.__clause_element__(), key)
475
476 def operate(
477 self, op: OperatorType, *other: Any, **kwargs: Any
478 ) -> ColumnElement[Any]:
479 return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
480
481 def reverse_operate(
482 self, op: OperatorType, other: Any, **kwargs: Any
483 ) -> ColumnElement[Any]:
484 col = self.__clause_element__()
485 return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501
486
487 def __str__(self) -> str:
488 if not self.parent or not self.key:
489 return object.__repr__(self)
490 return str(self.parent.class_.__name__) + "." + self.key
491
492
493class MappedSQLExpression(ColumnProperty[_T], _DeclarativeMapped[_T]):
494 """Declarative front-end for the :class:`.ColumnProperty` class.
495
496 Public constructor is the :func:`_orm.column_property` function.
497
498 .. versionchanged:: 2.0 Added :class:`_orm.MappedSQLExpression` as
499 a Declarative compatible subclass for :class:`_orm.ColumnProperty`.
500
501 .. seealso::
502
503 :class:`.MappedColumn`
504
505 """
506
507 inherit_cache = True
508 """:meta private:"""
509
510
511class MappedColumn(
512 _IntrospectsAnnotations,
513 _MapsColumns[_T],
514 _DeclarativeMapped[_T],
515):
516 """Maps a single :class:`_schema.Column` on a class.
517
518 :class:`_orm.MappedColumn` is a specialization of the
519 :class:`_orm.ColumnProperty` class and is oriented towards declarative
520 configuration.
521
522 To construct :class:`_orm.MappedColumn` objects, use the
523 :func:`_orm.mapped_column` constructor function.
524
525 .. versionadded:: 2.0
526
527
528 """
529
530 __slots__ = (
531 "column",
532 "_creation_order",
533 "_sort_order",
534 "foreign_keys",
535 "_has_nullable",
536 "_has_insert_default",
537 "deferred",
538 "deferred_group",
539 "deferred_raiseload",
540 "active_history",
541 "_attribute_options",
542 "_has_dataclass_arguments",
543 "_use_existing_column",
544 )
545
546 deferred: Union[_NoArg, bool]
547 deferred_raiseload: bool
548 deferred_group: Optional[str]
549
550 column: Column[_T]
551 foreign_keys: Optional[Set[ForeignKey]]
552 _attribute_options: _AttributeOptions
553
554 def __init__(self, *arg: Any, **kw: Any):
555 self._attribute_options = attr_opts = kw.pop(
556 "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS
557 )
558
559 self._use_existing_column = kw.pop("use_existing_column", False)
560
561 self._has_dataclass_arguments = (
562 attr_opts is not None
563 and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS
564 and any(
565 attr_opts[i] is not _NoArg.NO_ARG
566 for i, attr in enumerate(attr_opts._fields)
567 if attr != "dataclasses_default"
568 )
569 )
570
571 insert_default = kw.pop("insert_default", _NoArg.NO_ARG)
572 self._has_insert_default = insert_default is not _NoArg.NO_ARG
573
574 if self._has_insert_default:
575 kw["default"] = insert_default
576 elif attr_opts.dataclasses_default is not _NoArg.NO_ARG:
577 kw["default"] = attr_opts.dataclasses_default
578
579 self.deferred_group = kw.pop("deferred_group", None)
580 self.deferred_raiseload = kw.pop("deferred_raiseload", None)
581 self.deferred = kw.pop("deferred", _NoArg.NO_ARG)
582 self.active_history = kw.pop("active_history", False)
583
584 self._sort_order = kw.pop("sort_order", _NoArg.NO_ARG)
585 self.column = cast("Column[_T]", Column(*arg, **kw))
586 self.foreign_keys = self.column.foreign_keys
587 self._has_nullable = "nullable" in kw and kw.get("nullable") not in (
588 None,
589 SchemaConst.NULL_UNSPECIFIED,
590 )
591 util.set_creation_order(self)
592
593 def _copy(self, **kw: Any) -> Self:
594 new = self.__class__.__new__(self.__class__)
595 new.column = self.column._copy(**kw)
596 new.deferred = self.deferred
597 new.deferred_group = self.deferred_group
598 new.deferred_raiseload = self.deferred_raiseload
599 new.foreign_keys = new.column.foreign_keys
600 new.active_history = self.active_history
601 new._has_nullable = self._has_nullable
602 new._attribute_options = self._attribute_options
603 new._has_insert_default = self._has_insert_default
604 new._has_dataclass_arguments = self._has_dataclass_arguments
605 new._use_existing_column = self._use_existing_column
606 new._sort_order = self._sort_order
607 util.set_creation_order(new)
608 return new
609
610 @property
611 def name(self) -> str:
612 return self.column.name
613
614 @property
615 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]:
616 effective_deferred = self.deferred
617 if effective_deferred is _NoArg.NO_ARG:
618 effective_deferred = bool(
619 self.deferred_group or self.deferred_raiseload
620 )
621
622 if effective_deferred or self.active_history:
623 return ColumnProperty(
624 self.column,
625 deferred=effective_deferred,
626 group=self.deferred_group,
627 raiseload=self.deferred_raiseload,
628 attribute_options=self._attribute_options,
629 active_history=self.active_history,
630 )
631 else:
632 return None
633
634 @property
635 def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
636 return [
637 (
638 self.column,
639 (
640 self._sort_order
641 if self._sort_order is not _NoArg.NO_ARG
642 else 0
643 ),
644 )
645 ]
646
647 def __clause_element__(self) -> Column[_T]:
648 return self.column
649
650 def operate(
651 self, op: OperatorType, *other: Any, **kwargs: Any
652 ) -> ColumnElement[Any]:
653 return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
654
655 def reverse_operate(
656 self, op: OperatorType, other: Any, **kwargs: Any
657 ) -> ColumnElement[Any]:
658 col = self.__clause_element__()
659 return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501
660
661 def found_in_pep593_annotated(self) -> Any:
662 # return a blank mapped_column(). This mapped_column()'s
663 # Column will be merged into it in _init_column_for_annotation().
664 return MappedColumn()
665
666 def _adjust_for_existing_column(
667 self,
668 decl_scan: _ClassScanMapperConfig,
669 key: str,
670 given_column: Column[_T],
671 ) -> Column[_T]:
672 if (
673 self._use_existing_column
674 and decl_scan.inherits
675 and decl_scan.single
676 ):
677 if decl_scan.is_deferred:
678 raise sa_exc.ArgumentError(
679 "Can't use use_existing_column with deferred mappers"
680 )
681 supercls_mapper = class_mapper(decl_scan.inherits, False)
682
683 colname = (
684 given_column.name if given_column.name is not None else key
685 )
686 given_column = supercls_mapper.local_table.c.get( # type: ignore[assignment] # noqa: E501
687 colname, given_column
688 )
689 return given_column
690
691 def declarative_scan(
692 self,
693 decl_scan: _ClassScanMapperConfig,
694 registry: _RegistryType,
695 cls: Type[Any],
696 originating_module: Optional[str],
697 key: str,
698 mapped_container: Optional[Type[Mapped[Any]]],
699 annotation: Optional[_AnnotationScanType],
700 extracted_mapped_annotation: Optional[_AnnotationScanType],
701 is_dataclass_field: bool,
702 ) -> None:
703 column = self.column
704
705 column = self.column = self._adjust_for_existing_column(
706 decl_scan, key, self.column
707 )
708
709 if column.key is None:
710 column.key = key
711 if column.name is None:
712 column.name = key
713
714 sqltype = column.type
715
716 if extracted_mapped_annotation is None:
717 if sqltype._isnull and not self.column.foreign_keys:
718 self._raise_for_required(key, cls)
719 else:
720 return
721
722 self._init_column_for_annotation(
723 cls,
724 decl_scan,
725 key,
726 registry,
727 extracted_mapped_annotation,
728 originating_module,
729 )
730
731 @util.preload_module("sqlalchemy.orm.decl_base")
732 def declarative_scan_for_composite(
733 self,
734 decl_scan: _ClassScanMapperConfig,
735 registry: _RegistryType,
736 cls: Type[Any],
737 originating_module: Optional[str],
738 key: str,
739 param_name: str,
740 param_annotation: _AnnotationScanType,
741 ) -> None:
742 decl_base = util.preloaded.orm_decl_base
743 decl_base._undefer_column_name(param_name, self.column)
744 self._init_column_for_annotation(
745 cls, decl_scan, key, registry, param_annotation, originating_module
746 )
747
748 def _init_column_for_annotation(
749 self,
750 cls: Type[Any],
751 decl_scan: _ClassScanMapperConfig,
752 key: str,
753 registry: _RegistryType,
754 argument: _AnnotationScanType,
755 originating_module: Optional[str],
756 ) -> None:
757 sqltype = self.column.type
758
759 if is_fwd_ref(
760 argument, check_generic=True, check_for_plain_string=True
761 ):
762 assert originating_module is not None
763 argument = de_stringify_annotation(
764 cls, argument, originating_module, include_generic=True
765 )
766
767 nullable = includes_none(argument)
768
769 if not self._has_nullable:
770 self.column.nullable = nullable
771
772 find_mapped_in: Tuple[Any, ...] = ()
773 our_type_is_pep593 = False
774 raw_pep_593_type = None
775 raw_pep_695_type = None
776
777 our_type: Any = de_optionalize_union_types(argument)
778
779 if is_pep695(our_type):
780 raw_pep_695_type = our_type
781 our_type = de_optionalize_union_types(raw_pep_695_type.__value__)
782 our_args = get_args(raw_pep_695_type)
783 if our_args:
784 our_type = our_type[our_args]
785
786 if is_pep593(our_type):
787 our_type_is_pep593 = True
788
789 pep_593_components = get_args(our_type)
790 raw_pep_593_type = pep_593_components[0]
791 if nullable:
792 raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type)
793 find_mapped_in = pep_593_components[1:]
794
795 use_args_from: Optional[MappedColumn[Any]]
796 for elem in find_mapped_in:
797 if isinstance(elem, MappedColumn):
798 use_args_from = elem
799 break
800 else:
801 use_args_from = None
802
803 if use_args_from is not None:
804
805 self.column = use_args_from._adjust_for_existing_column(
806 decl_scan, key, self.column
807 )
808
809 if (
810 not self._has_insert_default
811 and use_args_from.column.default is not None
812 ):
813 self.column.default = None
814
815 use_args_from.column._merge(self.column)
816 sqltype = self.column.type
817
818 if (
819 use_args_from.deferred is not _NoArg.NO_ARG
820 and self.deferred is _NoArg.NO_ARG
821 ):
822 self.deferred = use_args_from.deferred
823
824 if (
825 use_args_from.deferred_group is not None
826 and self.deferred_group is None
827 ):
828 self.deferred_group = use_args_from.deferred_group
829
830 if (
831 use_args_from.deferred_raiseload is not None
832 and self.deferred_raiseload is None
833 ):
834 self.deferred_raiseload = use_args_from.deferred_raiseload
835
836 if (
837 use_args_from._use_existing_column
838 and not self._use_existing_column
839 ):
840 self._use_existing_column = True
841
842 if use_args_from.active_history:
843 self.active_history = use_args_from.active_history
844
845 if (
846 use_args_from._sort_order is not None
847 and self._sort_order is _NoArg.NO_ARG
848 ):
849 self._sort_order = use_args_from._sort_order
850
851 if (
852 use_args_from.column.key is not None
853 or use_args_from.column.name is not None
854 ):
855 util.warn_deprecated(
856 "Can't use the 'key' or 'name' arguments in "
857 "Annotated with mapped_column(); this will be ignored",
858 "2.0.22",
859 )
860
861 if use_args_from._has_dataclass_arguments:
862 for idx, arg in enumerate(
863 use_args_from._attribute_options._fields
864 ):
865 if (
866 use_args_from._attribute_options[idx]
867 is not _NoArg.NO_ARG
868 ):
869 arg = arg.replace("dataclasses_", "")
870 util.warn_deprecated(
871 f"Argument '{arg}' is a dataclass argument and "
872 "cannot be specified within a mapped_column() "
873 "bundled inside of an Annotated object",
874 "2.0.22",
875 )
876
877 if sqltype._isnull and not self.column.foreign_keys:
878 checks: List[Any]
879 if our_type_is_pep593:
880 checks = [our_type, raw_pep_593_type]
881 else:
882 checks = [our_type]
883
884 if raw_pep_695_type is not None:
885 checks.insert(0, raw_pep_695_type)
886
887 for check_type in checks:
888 new_sqltype = registry._resolve_type(
889 check_type, _do_fallbacks=check_type is our_type
890 )
891 if new_sqltype is not None:
892 break
893 else:
894 if isinstance(our_type, TypeEngine) or (
895 isinstance(our_type, type)
896 and issubclass(our_type, TypeEngine)
897 ):
898 raise orm_exc.MappedAnnotationError(
899 f"The type provided inside the {self.column.key!r} "
900 "attribute Mapped annotation is the SQLAlchemy type "
901 f"{our_type}. Expected a Python type instead"
902 )
903 elif is_a_type(checks[0]):
904 if len(checks) == 1:
905 detail = (
906 "the type object is not resolvable by the registry"
907 )
908 elif len(checks) == 2:
909 detail = (
910 f"neither '{checks[0]}' nor '{checks[1]}' "
911 "are resolvable by the registry"
912 )
913 else:
914 detail = (
915 f"""none of {
916 ", ".join(f"'{t}'" for t in checks)
917 } """
918 "are resolvable by the registry"
919 )
920 raise orm_exc.MappedAnnotationError(
921 "Could not locate SQLAlchemy Core type when resolving "
922 f"for Python type indicated by '{checks[0]}' inside "
923 "the "
924 f"Mapped[] annotation for the {self.column.key!r} "
925 f"attribute; {detail}"
926 )
927 else:
928 raise orm_exc.MappedAnnotationError(
929 f"The object provided inside the {self.column.key!r} "
930 "attribute Mapped annotation is not a Python type, "
931 f"it's the object {argument!r}. Expected a Python "
932 "type."
933 )
934
935 self.column._set_type(new_sqltype)
936