1# orm/decl_base.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"""Internal implementation for declarative."""
9
10from __future__ import annotations
11
12import collections
13import dataclasses
14import itertools
15import re
16from typing import Any
17from typing import Callable
18from typing import cast
19from typing import Dict
20from typing import get_args
21from typing import Iterable
22from typing import List
23from typing import Mapping
24from typing import NamedTuple
25from typing import NoReturn
26from typing import Optional
27from typing import Protocol
28from typing import Sequence
29from typing import Tuple
30from typing import Type
31from typing import TYPE_CHECKING
32from typing import TypeVar
33from typing import Union
34import weakref
35
36from . import attributes
37from . import clsregistry
38from . import exc as orm_exc
39from . import instrumentation
40from . import mapperlib
41from ._typing import _O
42from ._typing import attr_is_internal_proxy
43from .attributes import InstrumentedAttribute
44from .attributes import QueryableAttribute
45from .base import _is_mapped_class
46from .base import InspectionAttr
47from .descriptor_props import CompositeProperty
48from .descriptor_props import SynonymProperty
49from .interfaces import _AttributeOptions
50from .interfaces import _DataclassArguments
51from .interfaces import _DCAttributeOptions
52from .interfaces import _IntrospectsAnnotations
53from .interfaces import _MappedAttribute
54from .interfaces import _MapsColumns
55from .interfaces import MapperProperty
56from .mapper import Mapper
57from .properties import ColumnProperty
58from .properties import MappedColumn
59from .util import _extract_mapped_subtype
60from .util import _is_mapped_annotation
61from .util import _metadata_for_cls as _metadata_for_cls_fn
62from .util import class_mapper
63from .util import de_stringify_annotation
64from .. import event
65from .. import exc
66from .. import util
67from ..sql import expression
68from ..sql._annotated_cols import TypedColumns
69from ..sql.base import _NoArg
70from ..sql.schema import Column
71from ..sql.schema import Table
72from ..util import topological
73from ..util.typing import _AnnotationScanType
74from ..util.typing import is_fwd_ref
75from ..util.typing import is_literal
76
77if TYPE_CHECKING:
78 from ._typing import _ClassDict
79 from ._typing import _RegistryType
80 from .base import Mapped
81 from .decl_api import declared_attr
82 from .instrumentation import ClassManager
83 from ..sql.elements import NamedColumn
84 from ..sql.schema import MetaData
85 from ..sql.selectable import FromClause
86
87_T = TypeVar("_T", bound=Any)
88
89_MapperKwArgs = Mapping[str, Any]
90_TableArgsType = Union[Tuple[Any, ...], Dict[str, Any]]
91
92
93class MappedClassProtocol(Protocol[_O]):
94 """A protocol representing a SQLAlchemy mapped class.
95
96 The protocol is generic on the type of class, use
97 ``MappedClassProtocol[Any]`` to allow any mapped class.
98 """
99
100 __name__: str
101 __mapper__: Mapper[_O]
102 __table__: FromClause
103
104 def __call__(self, **kw: Any) -> _O: ...
105
106
107class _DeclMappedClassProtocol(MappedClassProtocol[_O], Protocol):
108 "Internal more detailed version of ``MappedClassProtocol``."
109
110 metadata: MetaData
111 __tablename__: str
112 __mapper_args__: _MapperKwArgs
113 __table_args__: Optional[_TableArgsType]
114
115 _sa_apply_dc_transforms: Optional[_DataclassArguments]
116
117 def __declare_first__(self) -> None: ...
118
119 def __declare_last__(self) -> None: ...
120
121
122def _declared_mapping_info(
123 cls: Type[Any],
124) -> Optional[Union[_DeferredDeclarativeConfig, Mapper[Any]]]:
125 # deferred mapping
126 if _DeferredDeclarativeConfig.has_cls(cls):
127 return _DeferredDeclarativeConfig.config_for_cls(cls)
128 # regular mapping
129 elif _is_mapped_class(cls):
130 return class_mapper(cls, configure=False)
131 else:
132 return None
133
134
135def _is_supercls_for_inherits(cls: Type[Any]) -> bool:
136 """return True if this class will be used as a superclass to set in
137 'inherits'.
138
139 This includes deferred mapper configs that aren't mapped yet, however does
140 not include classes with _sa_decl_prepare_nocascade (e.g.
141 ``AbstractConcreteBase``); these concrete-only classes are not set up as
142 "inherits" until after mappers are configured using
143 mapper._set_concrete_base()
144
145 """
146 if _DeferredDeclarativeConfig.has_cls(cls):
147 return not _get_immediate_cls_attr(
148 cls, "_sa_decl_prepare_nocascade", strict=True
149 )
150 # regular mapping
151 elif _is_mapped_class(cls):
152 return True
153 else:
154 return False
155
156
157def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
158 if cls is object:
159 return None
160
161 sup: Optional[Type[Any]]
162
163 if cls.__dict__.get("__abstract__", False):
164 for base_ in cls.__bases__:
165 sup = _resolve_for_abstract_or_classical(base_)
166 if sup is not None:
167 return sup
168 else:
169 return None
170 else:
171 clsmanager = _dive_for_cls_manager(cls)
172
173 if clsmanager:
174 return clsmanager.class_
175 else:
176 return cls
177
178
179def _get_immediate_cls_attr(
180 cls: Type[Any], attrname: str, strict: bool = False
181) -> Optional[Any]:
182 """return an attribute of the class that is either present directly
183 on the class, e.g. not on a superclass, or is from a superclass but
184 this superclass is a non-mapped mixin, that is, not a descendant of
185 the declarative base and is also not classically mapped.
186
187 This is used to detect attributes that indicate something about
188 a mapped class independently from any mapped classes that it may
189 inherit from.
190
191 """
192
193 # the rules are different for this name than others,
194 # make sure we've moved it out. transitional
195 assert attrname != "__abstract__"
196
197 if not issubclass(cls, object):
198 return None
199
200 if attrname in cls.__dict__:
201 return getattr(cls, attrname)
202
203 for base in cls.__mro__[1:]:
204 _is_classical_inherits = _dive_for_cls_manager(base) is not None
205
206 if attrname in base.__dict__ and (
207 base is cls
208 or (
209 (base in cls.__bases__ if strict else True)
210 and not _is_classical_inherits
211 )
212 ):
213 return getattr(base, attrname)
214 else:
215 return None
216
217
218def _dive_for_cls_manager(cls: Type[_O]) -> Optional[ClassManager[_O]]:
219 # because the class manager registration is pluggable,
220 # we need to do the search for every class in the hierarchy,
221 # rather than just a simple "cls._sa_class_manager"
222
223 for base in cls.__mro__:
224 manager: Optional[ClassManager[_O]] = attributes.opt_manager_of_class(
225 base
226 )
227 if manager:
228 return manager
229 return None
230
231
232@util.preload_module("sqlalchemy.orm.decl_api")
233def _is_declarative_props(obj: Any) -> bool:
234 _declared_attr_common = util.preloaded.orm_decl_api._declared_attr_common
235
236 return isinstance(obj, (_declared_attr_common, util.classproperty))
237
238
239def _check_declared_props_nocascade(
240 obj: Any, name: str, cls: Type[_O]
241) -> bool:
242 if _is_declarative_props(obj):
243 if getattr(obj, "_cascading", False):
244 util.warn(
245 "@declared_attr.cascading is not supported on the %s "
246 "attribute on class %s. This attribute invokes for "
247 "subclasses in any case." % (name, cls)
248 )
249 return True
250 else:
251 return False
252
253
254class _ORMClassConfigurator:
255 """Object that configures a class that's potentially going to be
256 mapped, and/or turned into an ORM dataclass.
257
258 This is the base class for all the configurator objects.
259
260 """
261
262 __slots__ = ("cls", "classname", "__weakref__")
263
264 cls: Type[Any]
265 classname: str
266
267 def __init__(self, cls_: Type[Any]):
268 self.cls = util.assert_arg_type(cls_, type, "cls_")
269 self.classname = cls_.__name__
270
271 @classmethod
272 def _as_declarative(
273 cls, registry: _RegistryType, cls_: Type[Any], dict_: _ClassDict
274 ) -> Optional[_MapperConfig]:
275 manager = attributes.opt_manager_of_class(cls_)
276 if manager and manager.class_ is cls_:
277 raise exc.InvalidRequestError(
278 f"Class {cls_!r} already has been instrumented declaratively"
279 )
280
281 # allow subclassing an orm class with typed columns without
282 # generating an orm class
283 if cls_.__dict__.get("__abstract__", False) or issubclass(
284 cls_, TypedColumns
285 ):
286 return None
287
288 defer_map = _get_immediate_cls_attr(
289 cls_, "_sa_decl_prepare_nocascade", strict=True
290 ) or hasattr(cls_, "_sa_decl_prepare")
291
292 if defer_map:
293 return _DeferredDeclarativeConfig(registry, cls_, dict_)
294 else:
295 return _DeclarativeMapperConfig(registry, cls_, dict_)
296
297 @classmethod
298 def _as_unmapped_dataclass(
299 cls, cls_: Type[Any], dict_: _ClassDict
300 ) -> _UnmappedDataclassConfig:
301 return _UnmappedDataclassConfig(cls_, dict_)
302
303 @classmethod
304 def _mapper(
305 cls,
306 registry: _RegistryType,
307 cls_: Type[_O],
308 table: Optional[FromClause],
309 mapper_kw: _MapperKwArgs,
310 ) -> Mapper[_O]:
311 _ImperativeMapperConfig(registry, cls_, table, mapper_kw)
312 return cast("MappedClassProtocol[_O]", cls_).__mapper__
313
314
315class _MapperConfig(_ORMClassConfigurator):
316 """Configurator that configures a class that's potentially going to be
317 mapped, and optionally turned into a dataclass as well."""
318
319 __slots__ = (
320 "properties",
321 "declared_attr_reg",
322 )
323
324 properties: util.OrderedDict[
325 str,
326 Union[
327 Sequence[NamedColumn[Any]], NamedColumn[Any], MapperProperty[Any]
328 ],
329 ]
330 declared_attr_reg: Dict[declared_attr[Any], Any]
331
332 def __init__(
333 self,
334 registry: _RegistryType,
335 cls_: Type[Any],
336 ):
337 super().__init__(cls_)
338 self.properties = util.OrderedDict()
339 self.declared_attr_reg = {}
340
341 instrumentation.register_class(
342 self.cls,
343 finalize=False,
344 registry=registry,
345 declarative_scan=self,
346 init_method=registry.constructor,
347 )
348
349 def set_cls_attribute(self, attrname: str, value: _T) -> _T:
350 manager = instrumentation.manager_of_class(self.cls)
351 manager.install_member(attrname, value)
352 return value
353
354 def map(self, mapper_kw: _MapperKwArgs) -> Mapper[Any]:
355 raise NotImplementedError()
356
357 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
358 self.map(mapper_kw)
359
360
361class _ImperativeMapperConfig(_MapperConfig):
362 """Configurator that configures a class for an imperative mapping."""
363
364 __slots__ = ("local_table", "inherits")
365
366 def __init__(
367 self,
368 registry: _RegistryType,
369 cls_: Type[_O],
370 table: Optional[FromClause],
371 mapper_kw: _MapperKwArgs,
372 ):
373 super().__init__(registry, cls_)
374
375 self.local_table = self.set_cls_attribute("__table__", table)
376
377 with mapperlib._CONFIGURE_MUTEX:
378 clsregistry._add_class(
379 self.classname, self.cls, registry._class_registry
380 )
381
382 self._setup_inheritance(mapper_kw)
383
384 self._early_mapping(mapper_kw)
385
386 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
387 mapper_cls = Mapper
388
389 return self.set_cls_attribute(
390 "__mapper__",
391 mapper_cls(self.cls, self.local_table, **mapper_kw),
392 )
393
394 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
395 cls = self.cls
396
397 inherits = None
398 inherits_search = []
399
400 # since we search for classical mappings now, search for
401 # multiple mapped bases as well and raise an error.
402 for base_ in cls.__bases__:
403 c = _resolve_for_abstract_or_classical(base_)
404 if c is None:
405 continue
406
407 if _is_supercls_for_inherits(c) and c not in inherits_search:
408 inherits_search.append(c)
409
410 if inherits_search:
411 if len(inherits_search) > 1:
412 raise exc.InvalidRequestError(
413 "Class %s has multiple mapped bases: %r"
414 % (cls, inherits_search)
415 )
416 inherits = inherits_search[0]
417
418 self.inherits = inherits
419
420
421class _CollectedAnnotation(NamedTuple):
422 raw_annotation: _AnnotationScanType
423 mapped_container: Optional[Type[Mapped[Any]]]
424 extracted_mapped_annotation: Union[_AnnotationScanType, str]
425 is_dataclass: bool
426 attr_value: Any
427 originating_module: str
428 originating_class: Type[Any]
429
430
431class _ClassScanAbstractConfig(_ORMClassConfigurator):
432 """Abstract base for a configurator that configures a class for a
433 declarative mapping, or an unmapped ORM dataclass.
434
435 Defines scanning of pep-484 annotations as well as ORM dataclass
436 applicators
437
438 """
439
440 __slots__ = ()
441
442 clsdict_view: _ClassDict
443 collected_annotations: Dict[str, _CollectedAnnotation]
444 collected_attributes: Dict[str, Any]
445
446 is_dataclass_prior_to_mapping: bool
447 allow_unmapped_annotations: bool
448
449 dataclass_setup_arguments: Optional[_DataclassArguments]
450 """if the class has SQLAlchemy native dataclass parameters, where
451 we will turn the class into a dataclass within the declarative mapping
452 process.
453
454 """
455
456 allow_dataclass_fields: bool
457 """if true, look for dataclass-processed Field objects on the target
458 class as well as superclasses and extract ORM mapping directives from
459 the "metadata" attribute of each Field.
460
461 if False, dataclass fields can still be used, however they won't be
462 mapped.
463
464 """
465
466 _include_dunders = {
467 "__table__",
468 "__mapper_args__",
469 "__tablename__",
470 "__table_args__",
471 }
472
473 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
474
475 def _scan_attributes(self) -> None:
476 raise NotImplementedError()
477
478 def _setup_dataclasses_transforms(
479 self, *, enable_descriptor_defaults: bool, revert: bool = False
480 ) -> None:
481 dataclass_setup_arguments = self.dataclass_setup_arguments
482 if not dataclass_setup_arguments:
483 return
484
485 # can't use is_dataclass since it uses hasattr
486 if "__dataclass_fields__" in self.cls.__dict__:
487 raise exc.InvalidRequestError(
488 f"Class {self.cls} is already a dataclass; ensure that "
489 "base classes / decorator styles of establishing dataclasses "
490 "are not being mixed. "
491 "This can happen if a class that inherits from "
492 "'MappedAsDataclass', even indirectly, is been mapped with "
493 "'@registry.mapped_as_dataclass'"
494 )
495
496 # can't create a dataclass if __table__ is already there. This would
497 # fail an assertion when calling _get_arguments_for_make_dataclass:
498 # assert False, "Mapped[] received without a mapping declaration"
499 if "__table__" in self.cls.__dict__:
500 raise exc.InvalidRequestError(
501 f"Class {self.cls} already defines a '__table__'. "
502 "ORM Annotated Dataclasses do not support a pre-existing "
503 "'__table__' element"
504 )
505
506 raise_for_non_dc_attrs = collections.defaultdict(list)
507
508 def _allow_dataclass_field(
509 key: str, originating_class: Type[Any]
510 ) -> bool:
511 if (
512 originating_class is not self.cls
513 and "__dataclass_fields__" not in originating_class.__dict__
514 ):
515 raise_for_non_dc_attrs[originating_class].append(key)
516
517 return True
518
519 field_list = [
520 _AttributeOptions._get_arguments_for_make_dataclass(
521 self,
522 key,
523 anno,
524 mapped_container,
525 self.collected_attributes.get(key, _NoArg.NO_ARG),
526 dataclass_setup_arguments,
527 enable_descriptor_defaults,
528 )
529 for key, anno, mapped_container in (
530 (
531 key,
532 raw_anno,
533 mapped_container,
534 )
535 for key, (
536 raw_anno,
537 mapped_container,
538 mapped_anno,
539 is_dc,
540 attr_value,
541 originating_module,
542 originating_class,
543 ) in self.collected_annotations.items()
544 if _allow_dataclass_field(key, originating_class)
545 and (
546 key not in self.collected_attributes
547 # issue #9226; check for attributes that we've collected
548 # which are already instrumented, which we would assume
549 # mean we are in an ORM inheritance mapping and this
550 # attribute is already mapped on the superclass. Under
551 # no circumstance should any QueryableAttribute be sent to
552 # the dataclass() function; anything that's mapped should
553 # be Field and that's it
554 or not isinstance(
555 self.collected_attributes[key], QueryableAttribute
556 )
557 )
558 )
559 ]
560 if raise_for_non_dc_attrs:
561 for (
562 originating_class,
563 non_dc_attrs,
564 ) in raise_for_non_dc_attrs.items():
565 raise exc.InvalidRequestError(
566 f"When transforming {self.cls} to a dataclass, "
567 f"attribute(s) "
568 f"{', '.join(repr(key) for key in non_dc_attrs)} "
569 f"originates from superclass "
570 f"{originating_class}, which is not a dataclass. When "
571 f"declaring SQLAlchemy Declarative "
572 f"Dataclasses, ensure that all mixin classes and other "
573 f"superclasses which include attributes are also a "
574 f"subclass of MappedAsDataclass or make use of the "
575 f"@unmapped_dataclass decorator.",
576 code="dcmx",
577 )
578
579 if revert:
580 # the "revert" case is used only by an unmapped mixin class
581 # that is nonetheless using Mapped construct and needs to
582 # itself be a dataclass
583 revert_dict = {
584 name: self.cls.__dict__[name]
585 for name in (item[0] for item in field_list)
586 if name in self.cls.__dict__
587 }
588 else:
589 revert_dict = None
590
591 # get original annotations using ForwardRef for symbols that
592 # are unresolvable
593 orig_annotations = util.get_annotations(self.cls)
594
595 # build a new __annotations__ dict from the fields we have.
596 # this has to be done carefully since we have to maintain
597 # the correct order! wow
598 swap_annotations = {}
599 defaults = {}
600
601 for item in field_list:
602 if len(item) == 2:
603 name, tp = item
604 elif len(item) == 3:
605 name, tp, spec = item
606 defaults[name] = spec
607 else:
608 assert False
609
610 # add the annotation to the new dict we are creating.
611 # note that if name is in orig_annotations, we expect
612 # tp and orig_annotations[name] to be identical.
613 swap_annotations[name] = orig_annotations.get(name, tp)
614
615 for k, v in defaults.items():
616 setattr(self.cls, k, v)
617
618 self._assert_dc_arguments(dataclass_setup_arguments)
619
620 dataclass_callable = dataclass_setup_arguments["dataclass_callable"]
621 if dataclass_callable is _NoArg.NO_ARG:
622 dataclass_callable = dataclasses.dataclass
623
624 # create a merged __annotations__ dictionary, maintaining order
625 # as best we can:
626
627 # 1. merge all keys in orig_annotations that occur before
628 # we see any of our mapped fields (this can be attributes like
629 # __table_args__ etc.)
630 new_annotations = {
631 k: orig_annotations[k]
632 for k in itertools.takewhile(
633 lambda k: k not in swap_annotations, orig_annotations
634 )
635 }
636
637 # 2. then put in all the dataclass annotations we have
638 new_annotations |= swap_annotations
639
640 # 3. them merge all of orig_annotations which will add remaining
641 # keys
642 new_annotations |= orig_annotations
643
644 # 4. this becomes the new class annotations.
645 restore_anno = util.restore_annotations(self.cls, new_annotations)
646
647 try:
648 dataclass_callable( # type: ignore[call-overload]
649 self.cls,
650 **{ # type: ignore[call-overload,unused-ignore]
651 k: v
652 for k, v in dataclass_setup_arguments.items()
653 if v is not _NoArg.NO_ARG
654 and k not in ("dataclass_callable",)
655 },
656 )
657 except (TypeError, ValueError) as ex:
658 raise exc.InvalidRequestError(
659 f"Python dataclasses error encountered when creating "
660 f"dataclass for {self.cls.__name__!r}: "
661 f"{ex!r}. Please refer to Python dataclasses "
662 "documentation for additional information.",
663 code="dcte",
664 ) from ex
665 else:
666 # as of Python 3.15, dataclasses no longer renders the
667 # auto-generated class docstring immediately; it instead installs
668 # a descriptor that renders the ``__init__`` signature the first
669 # time ``__doc__`` is accessed (see
670 # ``dataclasses._AutoDocstring``). As the ``finally:`` block
671 # below puts the class' original annotations back, that deferred
672 # render would no longer see the dataclass-oriented annotations
673 # we swapped in above, and would omit them from the docstring
674 # entirely. Read the attribute now, while those annotations are
675 # still in place, so the docstring we generate is the same on
676 # every Python version.
677 self.cls.__doc__
678
679 finally:
680 if revert and revert_dict:
681 # used for mixin dataclasses; we have to restore the
682 # mapped_column(), relationship() etc. to the class so these
683 # take place for a mapped class scan
684 for k, v in revert_dict.items():
685 setattr(self.cls, k, v)
686
687 restore_anno()
688
689 def _collect_annotation(
690 self,
691 name: str,
692 raw_annotation: _AnnotationScanType,
693 originating_class: Type[Any],
694 expect_mapped: Optional[bool],
695 attr_value: Any,
696 ) -> Optional[_CollectedAnnotation]:
697 if name in self.collected_annotations:
698 return self.collected_annotations[name]
699
700 if raw_annotation is None:
701 return None
702
703 is_dataclass = self.is_dataclass_prior_to_mapping
704 allow_unmapped = self.allow_unmapped_annotations
705
706 if expect_mapped is None:
707 is_dataclass_field = isinstance(attr_value, dataclasses.Field)
708 expect_mapped = (
709 not is_dataclass_field
710 and not allow_unmapped
711 and (
712 attr_value is None
713 or isinstance(attr_value, _MappedAttribute)
714 )
715 )
716
717 is_dataclass_field = False
718 extracted = _extract_mapped_subtype(
719 raw_annotation,
720 self.cls,
721 originating_class.__module__,
722 name,
723 type(attr_value),
724 required=False,
725 is_dataclass_field=is_dataclass_field,
726 expect_mapped=expect_mapped and not is_dataclass,
727 )
728 if extracted is None:
729 # ClassVar can come out here
730 return None
731
732 extracted_mapped_annotation, mapped_container = extracted
733
734 if attr_value is None and not is_literal(extracted_mapped_annotation):
735 for elem in get_args(extracted_mapped_annotation):
736 if is_fwd_ref(
737 elem, check_generic=True, check_for_plain_string=True
738 ):
739 elem = de_stringify_annotation(
740 self.cls,
741 elem,
742 originating_class.__module__,
743 include_generic=True,
744 )
745 # look in Annotated[...] for an ORM construct,
746 # such as Annotated[int, mapped_column(primary_key=True)]
747 if isinstance(elem, _IntrospectsAnnotations):
748 attr_value = elem.found_in_pep593_annotated()
749
750 self.collected_annotations[name] = ca = _CollectedAnnotation(
751 raw_annotation,
752 mapped_container,
753 extracted_mapped_annotation,
754 is_dataclass,
755 attr_value,
756 originating_class.__module__,
757 originating_class,
758 )
759 return ca
760
761 @classmethod
762 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None:
763 allowed = {
764 "init",
765 "repr",
766 "order",
767 "eq",
768 "unsafe_hash",
769 "kw_only",
770 "match_args",
771 "dataclass_callable",
772 }
773 disallowed_args = set(arguments).difference(allowed)
774 if disallowed_args:
775 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args))
776 raise exc.ArgumentError(
777 f"Dataclass argument(s) {msg} are not accepted"
778 )
779
780 def _cls_attr_override_checker(
781 self, cls: Type[_O]
782 ) -> Callable[[str, Any], bool]:
783 """Produce a function that checks if a class has overridden an
784 attribute, taking SQLAlchemy-enabled dataclass fields into account.
785
786 """
787
788 if self.allow_dataclass_fields:
789 sa_dataclass_metadata_key = _get_immediate_cls_attr(
790 cls, "__sa_dataclass_metadata_key__"
791 )
792 else:
793 sa_dataclass_metadata_key = None
794
795 if not sa_dataclass_metadata_key:
796
797 def attribute_is_overridden(key: str, obj: Any) -> bool:
798 return getattr(cls, key, obj) is not obj
799
800 else:
801 all_datacls_fields = {
802 f.name: f.metadata[sa_dataclass_metadata_key]
803 for f in util.dataclass_fields(cls)
804 if sa_dataclass_metadata_key in f.metadata
805 }
806 local_datacls_fields = {
807 f.name: f.metadata[sa_dataclass_metadata_key]
808 for f in util.local_dataclass_fields(cls)
809 if sa_dataclass_metadata_key in f.metadata
810 }
811
812 absent = object()
813
814 def attribute_is_overridden(key: str, obj: Any) -> bool:
815 if _is_declarative_props(obj):
816 obj = obj.fget
817
818 # this function likely has some failure modes still if
819 # someone is doing a deep mixing of the same attribute
820 # name as plain Python attribute vs. dataclass field.
821
822 ret = local_datacls_fields.get(key, absent)
823 if _is_declarative_props(ret):
824 ret = ret.fget
825
826 if ret is obj:
827 return False
828 elif ret is not absent:
829 return True
830
831 all_field = all_datacls_fields.get(key, absent)
832
833 ret = getattr(cls, key, obj)
834
835 if ret is obj:
836 return False
837
838 # for dataclasses, this could be the
839 # 'default' of the field. so filter more specifically
840 # for an already-mapped InstrumentedAttribute
841 if ret is not absent and isinstance(
842 ret, InstrumentedAttribute
843 ):
844 return True
845
846 if all_field is obj:
847 return False
848 elif all_field is not absent:
849 return True
850
851 # can't find another attribute
852 return False
853
854 return attribute_is_overridden
855
856 def _cls_attr_resolver(
857 self, cls: Type[Any]
858 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
859 """produce a function to iterate the "attributes" of a class
860 which we want to consider for mapping, adjusting for SQLAlchemy fields
861 embedded in dataclass fields.
862
863 """
864 cls_annotations = util.get_annotations(cls)
865
866 cls_vars = vars(cls)
867
868 _include_dunders = self._include_dunders
869 _match_exclude_dunders = self._match_exclude_dunders
870
871 names = [
872 n
873 for n in util.merge_lists_w_ordering(
874 list(cls_vars), list(cls_annotations)
875 )
876 if not _match_exclude_dunders.match(n) or n in _include_dunders
877 ]
878
879 if self.allow_dataclass_fields:
880 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
881 cls, "__sa_dataclass_metadata_key__"
882 )
883 else:
884 sa_dataclass_metadata_key = None
885
886 if not sa_dataclass_metadata_key:
887
888 def local_attributes_for_class() -> (
889 Iterable[Tuple[str, Any, Any, bool]]
890 ):
891 return (
892 (
893 name,
894 cls_vars.get(name),
895 cls_annotations.get(name),
896 False,
897 )
898 for name in names
899 )
900
901 else:
902 dataclass_fields = {
903 field.name: field for field in util.local_dataclass_fields(cls)
904 }
905
906 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
907
908 def local_attributes_for_class() -> (
909 Iterable[Tuple[str, Any, Any, bool]]
910 ):
911 for name in names:
912 field = dataclass_fields.get(name, None)
913 if field and sa_dataclass_metadata_key in field.metadata:
914 yield field.name, _as_dc_declaredattr(
915 field.metadata, fixed_sa_dataclass_metadata_key
916 ), cls_annotations.get(field.name), True
917 else:
918 yield name, cls_vars.get(name), cls_annotations.get(
919 name
920 ), False
921
922 return local_attributes_for_class
923
924
925class _DeclarativeMapperConfig(_MapperConfig, _ClassScanAbstractConfig):
926 """Configurator that will produce a declarative mapped class"""
927
928 __slots__ = (
929 "registry",
930 "local_table",
931 "persist_selectable",
932 "declared_columns",
933 "column_ordering",
934 "column_copies",
935 "table_args",
936 "tablename",
937 "mapper_args",
938 "mapper_args_fn",
939 "table_fn",
940 "inherits",
941 "single",
942 "clsdict_view",
943 "collected_attributes",
944 "collected_annotations",
945 "allow_dataclass_fields",
946 "dataclass_setup_arguments",
947 "is_dataclass_prior_to_mapping",
948 "allow_unmapped_annotations",
949 )
950
951 is_deferred = False
952 registry: _RegistryType
953 local_table: Optional[FromClause]
954 persist_selectable: Optional[FromClause]
955 declared_columns: util.OrderedSet[Column[Any]]
956 column_ordering: Dict[Column[Any], int]
957 column_copies: Dict[
958 Union[MappedColumn[Any], Column[Any]],
959 Union[MappedColumn[Any], Column[Any]],
960 ]
961 tablename: Optional[str]
962 mapper_args: Mapping[str, Any]
963 table_args: Optional[_TableArgsType]
964 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
965 inherits: Optional[Type[Any]]
966 single: bool
967
968 def __init__(
969 self,
970 registry: _RegistryType,
971 cls_: Type[_O],
972 dict_: _ClassDict,
973 ):
974 # grab class dict before the instrumentation manager has been added.
975 # reduces cycles
976 self.clsdict_view = (
977 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
978 )
979 super().__init__(registry, cls_)
980 self.registry = registry
981 self.persist_selectable = None
982
983 self.collected_attributes = {}
984 self.collected_annotations = {}
985 self.declared_columns = util.OrderedSet()
986 self.column_ordering = {}
987 self.column_copies = {}
988 self.single = False
989 self.dataclass_setup_arguments = dca = getattr(
990 self.cls, "_sa_apply_dc_transforms", None
991 )
992
993 self.allow_unmapped_annotations = getattr(
994 self.cls, "__allow_unmapped__", False
995 ) or bool(self.dataclass_setup_arguments)
996
997 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
998 cls_
999 )
1000
1001 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
1002
1003 # we don't want to consume Field objects from a not-already-dataclass.
1004 # the Field objects won't have their "name" or "type" populated,
1005 # and while it seems like we could just set these on Field as we
1006 # read them, Field is documented as "user read only" and we need to
1007 # stay far away from any off-label use of dataclasses APIs.
1008 if (not cld or dca) and sdk:
1009 raise exc.InvalidRequestError(
1010 "SQLAlchemy mapped dataclasses can't consume mapping "
1011 "information from dataclass.Field() objects if the immediate "
1012 "class is not already a dataclass."
1013 )
1014
1015 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
1016 # then also look inside of dataclass.Field() objects yielded by
1017 # dataclasses.get_fields(cls) when scanning for attributes
1018 self.allow_dataclass_fields = bool(sdk and cld)
1019
1020 self._setup_declared_events()
1021
1022 self._scan_attributes()
1023
1024 self._setup_dataclasses_transforms(enable_descriptor_defaults=True)
1025
1026 with mapperlib._CONFIGURE_MUTEX:
1027 clsregistry._add_class(
1028 self.classname, self.cls, registry._class_registry
1029 )
1030
1031 self._setup_inheriting_mapper()
1032
1033 self._extract_mappable_attributes()
1034
1035 self._extract_declared_columns()
1036
1037 self._setup_table()
1038
1039 self._setup_inheriting_columns()
1040
1041 self._early_mapping(util.EMPTY_DICT)
1042
1043 def _setup_declared_events(self) -> None:
1044 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
1045
1046 @event.listens_for(Mapper, "after_configured")
1047 def after_configured() -> None:
1048 cast(
1049 "_DeclMappedClassProtocol[Any]", self.cls
1050 ).__declare_last__()
1051
1052 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
1053
1054 @event.listens_for(Mapper, "before_configured")
1055 def before_configured() -> None:
1056 cast(
1057 "_DeclMappedClassProtocol[Any]", self.cls
1058 ).__declare_first__()
1059
1060 def _scan_attributes(self) -> None:
1061 cls = self.cls
1062
1063 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
1064
1065 clsdict_view = self.clsdict_view
1066 collected_attributes = self.collected_attributes
1067 column_copies = self.column_copies
1068 _include_dunders = self._include_dunders
1069 mapper_args_fn = None
1070 table_args = inherited_table_args = None
1071 table_fn = None
1072 tablename = None
1073 fixed_table = "__table__" in clsdict_view
1074
1075 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
1076
1077 bases = []
1078
1079 for base in cls.__mro__:
1080 # collect bases and make sure standalone columns are copied
1081 # to be the column they will ultimately be on the class,
1082 # so that declared_attr functions use the right columns.
1083 # need to do this all the way up the hierarchy first
1084 # (see #8190)
1085
1086 class_mapped = base is not cls and _is_supercls_for_inherits(base)
1087
1088 local_attributes_for_class = self._cls_attr_resolver(base)
1089
1090 if not class_mapped and base is not cls:
1091 locally_collected_columns = self._produce_column_copies(
1092 local_attributes_for_class,
1093 attribute_is_overridden,
1094 fixed_table,
1095 base,
1096 )
1097 else:
1098 locally_collected_columns = {}
1099
1100 bases.append(
1101 (
1102 base,
1103 class_mapped,
1104 local_attributes_for_class,
1105 locally_collected_columns,
1106 )
1107 )
1108
1109 for (
1110 base,
1111 class_mapped,
1112 local_attributes_for_class,
1113 locally_collected_columns,
1114 ) in bases:
1115 # this transfer can also take place as we scan each name
1116 # for finer-grained control of how collected_attributes is
1117 # populated, as this is what impacts column ordering.
1118 # however it's simpler to get it out of the way here.
1119 collected_attributes.update(locally_collected_columns)
1120
1121 for (
1122 name,
1123 obj,
1124 annotation,
1125 is_dataclass_field,
1126 ) in local_attributes_for_class():
1127 if name in _include_dunders:
1128 if name == "__mapper_args__":
1129 check_decl = _check_declared_props_nocascade(
1130 obj, name, cls
1131 )
1132 if not mapper_args_fn and (
1133 not class_mapped or check_decl
1134 ):
1135 # don't even invoke __mapper_args__ until
1136 # after we've determined everything about the
1137 # mapped table.
1138 # make a copy of it so a class-level dictionary
1139 # is not overwritten when we update column-based
1140 # arguments.
1141 def _mapper_args_fn() -> Dict[str, Any]:
1142 return dict(cls_as_Decl.__mapper_args__)
1143
1144 mapper_args_fn = _mapper_args_fn
1145
1146 elif name == "__tablename__":
1147 check_decl = _check_declared_props_nocascade(
1148 obj, name, cls
1149 )
1150 if not tablename and (not class_mapped or check_decl):
1151 tablename = cls_as_Decl.__tablename__
1152 elif name == "__table__":
1153 check_decl = _check_declared_props_nocascade(
1154 obj, name, cls
1155 )
1156 # if a @declared_attr using "__table__" is detected,
1157 # wrap up a callable to look for "__table__" from
1158 # the final concrete class when we set up a table.
1159 # this was fixed by
1160 # #11509, regression in 2.0 from version 1.4.
1161 if check_decl and not table_fn:
1162 # don't even invoke __table__ until we're ready
1163 def _table_fn() -> FromClause:
1164 return cls_as_Decl.__table__
1165
1166 table_fn = _table_fn
1167
1168 elif name == "__table_args__":
1169 check_decl = _check_declared_props_nocascade(
1170 obj, name, cls
1171 )
1172 if not table_args and (not class_mapped or check_decl):
1173 table_args = cls_as_Decl.__table_args__
1174 if not isinstance(
1175 table_args, (tuple, dict, type(None))
1176 ):
1177 raise exc.ArgumentError(
1178 "__table_args__ value must be a tuple, "
1179 "dict, or None"
1180 )
1181 if base is not cls:
1182 inherited_table_args = True
1183 else:
1184 # any other dunder names; should not be here
1185 # as we have tested for all four names in
1186 # _include_dunders
1187 assert False
1188 elif class_mapped:
1189 if _is_declarative_props(obj) and not obj._quiet:
1190 util.warn(
1191 "Regular (i.e. not __special__) "
1192 "attribute '%s.%s' uses @declared_attr, "
1193 "but owning class %s is mapped - "
1194 "not applying to subclass %s."
1195 % (base.__name__, name, base, cls)
1196 )
1197
1198 continue
1199 elif base is not cls:
1200 # we're a mixin, abstract base, or something that is
1201 # acting like that for now.
1202
1203 if isinstance(obj, (Column, MappedColumn)):
1204 # already copied columns to the mapped class.
1205 continue
1206 elif isinstance(obj, MapperProperty):
1207 raise exc.InvalidRequestError(
1208 "Mapper properties (i.e. deferred,"
1209 "column_property(), relationship(), etc.) must "
1210 "be declared as @declared_attr callables "
1211 "on declarative mixin classes. For dataclass "
1212 "field() objects, use a lambda:"
1213 )
1214 elif _is_declarative_props(obj):
1215 # tried to get overloads to tell this to
1216 # pylance, no luck
1217 assert obj is not None
1218
1219 if obj._cascading:
1220 if name in clsdict_view:
1221 # unfortunately, while we can use the user-
1222 # defined attribute here to allow a clean
1223 # override, if there's another
1224 # subclass below then it still tries to use
1225 # this. not sure if there is enough
1226 # information here to add this as a feature
1227 # later on.
1228 util.warn(
1229 "Attribute '%s' on class %s cannot be "
1230 "processed due to "
1231 "@declared_attr.cascading; "
1232 "skipping" % (name, cls)
1233 )
1234 collected_attributes[name] = column_copies[obj] = (
1235 ret
1236 ) = obj.__get__(obj, cls)
1237 setattr(cls, name, ret)
1238 else:
1239 if is_dataclass_field:
1240 # access attribute using normal class access
1241 # first, to see if it's been mapped on a
1242 # superclass. note if the dataclasses.field()
1243 # has "default", this value can be anything.
1244 ret = getattr(cls, name, None)
1245
1246 # so, if it's anything that's not ORM
1247 # mapped, assume we should invoke the
1248 # declared_attr
1249 if not isinstance(ret, InspectionAttr):
1250 ret = obj.fget()
1251 else:
1252 # access attribute using normal class access.
1253 # if the declared attr already took place
1254 # on a superclass that is mapped, then
1255 # this is no longer a declared_attr, it will
1256 # be the InstrumentedAttribute
1257 ret = getattr(cls, name)
1258
1259 # correct for proxies created from hybrid_property
1260 # or similar. note there is no known case that
1261 # produces nested proxies, so we are only
1262 # looking one level deep right now.
1263
1264 if (
1265 isinstance(ret, InspectionAttr)
1266 and attr_is_internal_proxy(ret)
1267 and not isinstance(
1268 ret.original_property, MapperProperty
1269 )
1270 ):
1271 ret = ret.descriptor
1272
1273 collected_attributes[name] = column_copies[obj] = (
1274 ret
1275 )
1276
1277 if (
1278 isinstance(ret, (Column, MapperProperty))
1279 and ret.doc is None
1280 ):
1281 ret.doc = obj.__doc__
1282
1283 self._collect_annotation(
1284 name,
1285 obj._collect_return_annotation(),
1286 base,
1287 True,
1288 obj,
1289 )
1290 elif _is_mapped_annotation(annotation, cls, base):
1291 # Mapped annotation without any object.
1292 # product_column_copies should have handled this.
1293 # if future support for other MapperProperty,
1294 # then test if this name is already handled and
1295 # otherwise proceed to generate.
1296 if not fixed_table:
1297 assert (
1298 name in collected_attributes
1299 or attribute_is_overridden(name, None)
1300 )
1301 continue
1302 else:
1303 # here, the attribute is some other kind of
1304 # property that we assume is not part of the
1305 # declarative mapping. however, check for some
1306 # more common mistakes
1307 self._warn_for_decl_attributes(base, name, obj)
1308 elif is_dataclass_field and (
1309 name not in clsdict_view or clsdict_view[name] is not obj
1310 ):
1311 # here, we are definitely looking at the target class
1312 # and not a superclass. this is currently a
1313 # dataclass-only path. if the name is only
1314 # a dataclass field and isn't in local cls.__dict__,
1315 # put the object there.
1316 # assert that the dataclass-enabled resolver agrees
1317 # with what we are seeing
1318
1319 assert not attribute_is_overridden(name, obj)
1320
1321 if _is_declarative_props(obj):
1322 obj = obj.fget()
1323
1324 collected_attributes[name] = obj
1325 self._collect_annotation(
1326 name, annotation, base, False, obj
1327 )
1328 else:
1329 collected_annotation = self._collect_annotation(
1330 name, annotation, base, None, obj
1331 )
1332 is_mapped = (
1333 collected_annotation is not None
1334 and collected_annotation.mapped_container is not None
1335 )
1336 generated_obj = (
1337 collected_annotation.attr_value
1338 if collected_annotation is not None
1339 else obj
1340 )
1341 if obj is None and not fixed_table and is_mapped:
1342 collected_attributes[name] = (
1343 generated_obj
1344 if generated_obj is not None
1345 else MappedColumn()
1346 )
1347 elif name in clsdict_view:
1348 collected_attributes[name] = obj
1349 # else if the name is not in the cls.__dict__,
1350 # don't collect it as an attribute.
1351 # we will see the annotation only, which is meaningful
1352 # both for mapping and dataclasses setup
1353
1354 if inherited_table_args and not tablename:
1355 table_args = None
1356
1357 self.table_args = table_args
1358 self.tablename = tablename
1359 self.mapper_args_fn = mapper_args_fn
1360 self.table_fn = table_fn
1361
1362 @classmethod
1363 def _update_annotations_for_non_mapped_class(
1364 cls, klass: Type[_O]
1365 ) -> Mapping[str, _AnnotationScanType]:
1366 cls_annotations = util.get_annotations(klass)
1367
1368 new_anno = {}
1369 for name, annotation in cls_annotations.items():
1370 if _is_mapped_annotation(annotation, klass, klass):
1371 extracted = _extract_mapped_subtype(
1372 annotation,
1373 klass,
1374 klass.__module__,
1375 name,
1376 type(None),
1377 required=False,
1378 is_dataclass_field=False,
1379 expect_mapped=False,
1380 )
1381 if extracted:
1382 inner, _ = extracted
1383 new_anno[name] = inner
1384 else:
1385 new_anno[name] = annotation
1386 return new_anno
1387
1388 def _warn_for_decl_attributes(
1389 self, cls: Type[Any], key: str, c: Any
1390 ) -> None:
1391 if isinstance(c, expression.ColumnElement):
1392 util.warn(
1393 f"Attribute '{key}' on class {cls} appears to "
1394 "be a non-schema SQLAlchemy expression "
1395 "object; this won't be part of the declarative mapping. "
1396 "To map arbitrary expressions, use ``column_property()`` "
1397 "or a similar function such as ``deferred()``, "
1398 "``query_expression()`` etc. "
1399 )
1400
1401 def _produce_column_copies(
1402 self,
1403 attributes_for_class: Callable[
1404 [], Iterable[Tuple[str, Any, Any, bool]]
1405 ],
1406 attribute_is_overridden: Callable[[str, Any], bool],
1407 fixed_table: bool,
1408 originating_class: Type[Any],
1409 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]:
1410 cls = self.cls
1411 dict_ = self.clsdict_view
1412 locally_collected_attributes = {}
1413 column_copies = self.column_copies
1414 # copy mixin columns to the mapped class
1415
1416 for name, obj, annotation, is_dataclass in attributes_for_class():
1417 if (
1418 not fixed_table
1419 and obj is None
1420 and _is_mapped_annotation(annotation, cls, originating_class)
1421 ):
1422 # obj is None means this is the annotation only path
1423
1424 if attribute_is_overridden(name, obj):
1425 # perform same "overridden" check as we do for
1426 # Column/MappedColumn, this is how a mixin col is not
1427 # applied to an inherited subclass that does not have
1428 # the mixin. the anno-only path added here for
1429 # #9564
1430 continue
1431
1432 collected_annotation = self._collect_annotation(
1433 name, annotation, originating_class, True, obj
1434 )
1435 obj = (
1436 collected_annotation.attr_value
1437 if collected_annotation is not None
1438 else obj
1439 )
1440 if obj is None:
1441 obj = MappedColumn()
1442
1443 locally_collected_attributes[name] = obj
1444 setattr(cls, name, obj)
1445
1446 elif isinstance(obj, (Column, MappedColumn)):
1447 if attribute_is_overridden(name, obj):
1448 # if column has been overridden
1449 # (like by the InstrumentedAttribute of the
1450 # superclass), skip. don't collect the annotation
1451 # either (issue #8718)
1452 continue
1453
1454 collected_annotation = self._collect_annotation(
1455 name, annotation, originating_class, True, obj
1456 )
1457 obj = (
1458 collected_annotation.attr_value
1459 if collected_annotation is not None
1460 else obj
1461 )
1462
1463 if name not in dict_ and not (
1464 "__table__" in dict_
1465 and (getattr(obj, "name", None) or name)
1466 in dict_["__table__"].c
1467 ):
1468 if obj.foreign_keys:
1469 for fk in obj.foreign_keys:
1470 if (
1471 fk._table_column is not None
1472 and fk._table_column.table is None
1473 ):
1474 raise exc.InvalidRequestError(
1475 "Columns with foreign keys to "
1476 "non-table-bound "
1477 "columns must be declared as "
1478 "@declared_attr callables "
1479 "on declarative mixin classes. "
1480 "For dataclass "
1481 "field() objects, use a lambda:."
1482 )
1483
1484 column_copies[obj] = copy_ = obj._copy()
1485
1486 locally_collected_attributes[name] = copy_
1487 setattr(cls, name, copy_)
1488
1489 return locally_collected_attributes
1490
1491 def _extract_mappable_attributes(self) -> None:
1492 cls = self.cls
1493 collected_attributes = self.collected_attributes
1494
1495 our_stuff = self.properties
1496
1497 _include_dunders = self._include_dunders
1498
1499 late_mapped = _get_immediate_cls_attr(
1500 cls, "_sa_decl_prepare_nocascade", strict=True
1501 )
1502
1503 allow_unmapped_annotations = self.allow_unmapped_annotations
1504 expect_annotations_wo_mapped = (
1505 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping
1506 )
1507
1508 look_for_dataclass_things = bool(self.dataclass_setup_arguments)
1509
1510 for k in list(collected_attributes):
1511 if k in _include_dunders:
1512 continue
1513
1514 value = collected_attributes[k]
1515
1516 if _is_declarative_props(value):
1517 # @declared_attr in collected_attributes only occurs here for a
1518 # @declared_attr that's directly on the mapped class;
1519 # for a mixin, these have already been evaluated
1520 if value._cascading:
1521 util.warn(
1522 "Use of @declared_attr.cascading only applies to "
1523 "Declarative 'mixin' and 'abstract' classes. "
1524 "Currently, this flag is ignored on mapped class "
1525 "%s" % self.cls
1526 )
1527
1528 value = getattr(cls, k)
1529
1530 elif (
1531 isinstance(value, QueryableAttribute)
1532 and value.class_ is not cls
1533 and value.key != k
1534 ):
1535 # detect a QueryableAttribute that's already mapped being
1536 # assigned elsewhere in userland, turn into a synonym()
1537 value = SynonymProperty(value.key)
1538 setattr(cls, k, value)
1539
1540 if k in ("metadata", "registry"):
1541 noun = {
1542 "metadata": "MetaData instance",
1543 "registry": "Declarative class registry",
1544 }
1545 util.warn(
1546 f"Attribute name '{k}' should be left reserved for the "
1547 f"{noun[k]} when using a Declarative class."
1548 )
1549
1550 if (
1551 isinstance(value, tuple)
1552 and len(value) == 1
1553 and isinstance(value[0], (Column, _MappedAttribute))
1554 ):
1555 util.warn(
1556 "Ignoring declarative-like tuple value of attribute "
1557 "'%s': possibly a copy-and-paste error with a comma "
1558 "accidentally placed at the end of the line?" % k
1559 )
1560 continue
1561 elif look_for_dataclass_things and isinstance(
1562 value, dataclasses.Field
1563 ):
1564 # we collected a dataclass Field; dataclasses would have
1565 # set up the correct state on the class
1566 continue
1567 elif not isinstance(value, (Column, _DCAttributeOptions)):
1568 # using @declared_attr for some object that
1569 # isn't Column/MapperProperty/_DCAttributeOptions; remove
1570 # from the clsdict_view
1571 # and place the evaluated value onto the class.
1572 collected_attributes.pop(k)
1573 self._warn_for_decl_attributes(cls, k, value)
1574 if not late_mapped:
1575 setattr(cls, k, value)
1576 continue
1577 elif isinstance(value, Column):
1578 _undefer_column_name(
1579 k, self.column_copies.get(value, value) # type: ignore[arg-type] # noqa: E501
1580 )
1581 else:
1582 if isinstance(value, _IntrospectsAnnotations):
1583 (
1584 annotation,
1585 mapped_container,
1586 extracted_mapped_annotation,
1587 is_dataclass,
1588 attr_value,
1589 originating_module,
1590 originating_class,
1591 ) = self.collected_annotations.get(
1592 k, (None, None, None, False, None, None, None)
1593 )
1594
1595 # issue #8692 - don't do any annotation interpretation if
1596 # an annotation were present and a container such as
1597 # Mapped[] etc. were not used. If annotation is None,
1598 # do declarative_scan so that the property can raise
1599 # for required
1600 if (
1601 mapped_container is not None
1602 or annotation is None
1603 # issue #10516: need to do declarative_scan even with
1604 # a non-Mapped annotation if we are doing
1605 # __allow_unmapped__, for things like col.name
1606 # assignment
1607 or allow_unmapped_annotations
1608 ):
1609 try:
1610 value.declarative_scan(
1611 self,
1612 self.registry,
1613 cls,
1614 originating_module,
1615 k,
1616 mapped_container,
1617 annotation,
1618 extracted_mapped_annotation,
1619 is_dataclass,
1620 )
1621 except NameError as ne:
1622 raise orm_exc.MappedAnnotationError(
1623 f"Could not resolve all types within mapped "
1624 f'annotation: "{annotation}". Ensure all '
1625 f"types are written correctly and are "
1626 f"imported within the module in use."
1627 ) from ne
1628 else:
1629 # assert that we were expecting annotations
1630 # without Mapped[] were going to be passed.
1631 # otherwise an error should have been raised
1632 # by util._extract_mapped_subtype before we got here.
1633 assert expect_annotations_wo_mapped
1634
1635 if isinstance(value, _DCAttributeOptions):
1636 if (
1637 value._has_dataclass_arguments
1638 and not look_for_dataclass_things
1639 ):
1640 if isinstance(value, MapperProperty):
1641 argnames = [
1642 "init",
1643 "default_factory",
1644 "repr",
1645 "default",
1646 "dataclass_metadata",
1647 ]
1648 else:
1649 argnames = [
1650 "init",
1651 "default_factory",
1652 "repr",
1653 "dataclass_metadata",
1654 ]
1655
1656 args = {
1657 a
1658 for a in argnames
1659 if getattr(
1660 value._attribute_options, f"dataclasses_{a}"
1661 )
1662 is not _NoArg.NO_ARG
1663 }
1664
1665 raise exc.ArgumentError(
1666 f"Attribute '{k}' on class {cls} includes "
1667 f"dataclasses argument(s): "
1668 f"{', '.join(sorted(repr(a) for a in args))} but "
1669 f"class does not specify "
1670 "SQLAlchemy native dataclass configuration."
1671 )
1672
1673 if not isinstance(value, (MapperProperty, _MapsColumns)):
1674 # filter for _DCAttributeOptions objects that aren't
1675 # MapperProperty / mapped_column(). Currently this
1676 # includes AssociationProxy. pop it from the things
1677 # we're going to map and set it up as a descriptor
1678 # on the class.
1679 collected_attributes.pop(k)
1680
1681 # Assoc Prox (or other descriptor object that may
1682 # use _DCAttributeOptions) is usually here, except if
1683 # 1. we're a
1684 # dataclass, dataclasses would have removed the
1685 # attr here or 2. assoc proxy is coming from a
1686 # superclass, we want it to be direct here so it
1687 # tracks state or 3. assoc prox comes from
1688 # declared_attr, uncommon case
1689 setattr(cls, k, value)
1690 continue
1691
1692 our_stuff[k] = value
1693
1694 def _extract_declared_columns(self) -> None:
1695 our_stuff = self.properties
1696
1697 # extract columns from the class dict
1698 declared_columns = self.declared_columns
1699 column_ordering = self.column_ordering
1700 name_to_prop_key = collections.defaultdict(set)
1701
1702 for key, c in list(our_stuff.items()):
1703 if isinstance(c, _MapsColumns):
1704 mp_to_assign = c.mapper_property_to_assign
1705 if mp_to_assign:
1706 our_stuff[key] = mp_to_assign
1707 else:
1708 # if no mapper property to assign, this currently means
1709 # this is a MappedColumn that will produce a Column for us
1710 del our_stuff[key]
1711
1712 for col, sort_order in c.columns_to_assign:
1713 if not isinstance(c, CompositeProperty):
1714 name_to_prop_key[col.name].add(key)
1715 declared_columns.add(col)
1716
1717 # we would assert this, however we want the below
1718 # warning to take effect instead. See #9630
1719 # assert col not in column_ordering
1720
1721 column_ordering[col] = sort_order
1722
1723 # if this is a MappedColumn and the attribute key we
1724 # have is not what the column has for its key, map the
1725 # Column explicitly under the attribute key name.
1726 # otherwise, Mapper will map it under the column key.
1727 if mp_to_assign is None and key != col.key:
1728 our_stuff[key] = col
1729 elif isinstance(c, Column):
1730 # undefer previously occurred here, and now occurs earlier.
1731 # ensure every column we get here has been named
1732 assert c.name is not None
1733 name_to_prop_key[c.name].add(key)
1734 declared_columns.add(c)
1735 # if the column is the same name as the key,
1736 # remove it from the explicit properties dict.
1737 # the normal rules for assigning column-based properties
1738 # will take over, including precedence of columns
1739 # in multi-column ColumnProperties.
1740 if key == c.key:
1741 del our_stuff[key]
1742
1743 for name, keys in name_to_prop_key.items():
1744 if len(keys) > 1:
1745 util.warn(
1746 "On class %r, Column object %r named "
1747 "directly multiple times, "
1748 "only one will be used: %s. "
1749 "Consider using orm.synonym instead"
1750 % (self.classname, name, (", ".join(sorted(keys))))
1751 )
1752
1753 def _setup_table(self, table: Optional[FromClause] = None) -> None:
1754 cls = self.cls
1755 cls_as_Decl = cast("MappedClassProtocol[Any]", cls)
1756
1757 tablename = self.tablename
1758 table_args = self.table_args
1759 clsdict_view = self.clsdict_view
1760 declared_columns = self.declared_columns
1761 column_ordering = self.column_ordering
1762
1763 manager = attributes.manager_of_class(cls)
1764
1765 if (
1766 self.table_fn is None
1767 and "__table__" not in clsdict_view
1768 and table is None
1769 ):
1770 if hasattr(cls, "__table_cls__"):
1771 table_cls = cast(
1772 Type[Table],
1773 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore[no-untyped-call] # noqa: E501
1774 )
1775 else:
1776 table_cls = Table
1777
1778 if tablename is not None:
1779 args: Tuple[Any, ...] = ()
1780 table_kw: Dict[str, Any] = {}
1781
1782 if table_args:
1783 if isinstance(table_args, dict):
1784 table_kw = table_args
1785 elif isinstance(table_args, tuple):
1786 if isinstance(table_args[-1], dict):
1787 args, table_kw = table_args[0:-1], table_args[-1]
1788 else:
1789 args = table_args
1790
1791 autoload_with = clsdict_view.get("__autoload_with__")
1792 if autoload_with:
1793 table_kw["autoload_with"] = autoload_with
1794
1795 autoload = clsdict_view.get("__autoload__")
1796 if autoload:
1797 table_kw["autoload"] = True
1798
1799 sorted_columns = sorted(
1800 declared_columns,
1801 key=lambda c: column_ordering.get(c, 0),
1802 )
1803 table = self.set_cls_attribute(
1804 "__table__",
1805 table_cls(
1806 tablename,
1807 self._metadata_for_cls(manager),
1808 *sorted_columns,
1809 *args,
1810 **table_kw,
1811 ),
1812 )
1813 else:
1814 if table is None:
1815 if self.table_fn:
1816 table = self.set_cls_attribute(
1817 "__table__", self.table_fn()
1818 )
1819 else:
1820 table = cls_as_Decl.__table__
1821 if declared_columns:
1822 for c in declared_columns:
1823 if not table.c.contains_column(c):
1824 raise exc.ArgumentError(
1825 "Can't add additional column %r when "
1826 "specifying __table__" % c.key
1827 )
1828
1829 self.local_table = table
1830
1831 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData:
1832 return _metadata_for_cls_fn(self.cls, manager.registry)
1833
1834 def _setup_inheriting_mapper(self) -> None:
1835 cls = self.cls
1836
1837 inherits = None
1838
1839 if inherits is None:
1840 # since we search for classical mappings now, search for
1841 # multiple mapped bases as well and raise an error.
1842 inherits_search = []
1843 for base_ in cls.__bases__:
1844 c = _resolve_for_abstract_or_classical(base_)
1845 if c is None:
1846 continue
1847
1848 if _is_supercls_for_inherits(c) and c not in inherits_search:
1849 inherits_search.append(c)
1850
1851 if inherits_search:
1852 if len(inherits_search) > 1:
1853 raise exc.InvalidRequestError(
1854 "Class %s has multiple mapped bases: %r"
1855 % (cls, inherits_search)
1856 )
1857 inherits = inherits_search[0]
1858 elif isinstance(inherits, Mapper):
1859 inherits = inherits.class_
1860
1861 self.inherits = inherits
1862
1863 clsdict_view = self.clsdict_view
1864 if "__table__" not in clsdict_view and self.tablename is None:
1865 self.single = True
1866
1867 def _setup_inheriting_columns(self) -> None:
1868 table = self.local_table
1869 cls = self.cls
1870 table_args = self.table_args
1871 declared_columns = self.declared_columns
1872
1873 if (
1874 table is None
1875 and self.inherits is None
1876 and not _get_immediate_cls_attr(cls, "__no_table__")
1877 ):
1878 raise exc.InvalidRequestError(
1879 "Class %r does not have a __table__ or __tablename__ "
1880 "specified and does not inherit from an existing "
1881 "table-mapped class." % cls
1882 )
1883 elif self.inherits:
1884 inherited_mapper_or_config = _declared_mapping_info(self.inherits)
1885 assert inherited_mapper_or_config is not None
1886 inherited_table = inherited_mapper_or_config.local_table
1887 inherited_persist_selectable = (
1888 inherited_mapper_or_config.persist_selectable
1889 )
1890
1891 if table is None:
1892 # single table inheritance.
1893 # ensure no table args
1894 if table_args:
1895 raise exc.ArgumentError(
1896 "Can't place __table_args__ on an inherited class "
1897 "with no table."
1898 )
1899
1900 # add any columns declared here to the inherited table.
1901 if declared_columns and not isinstance(inherited_table, Table):
1902 raise exc.ArgumentError(
1903 f"Can't declare columns on single-table-inherited "
1904 f"subclass {self.cls}; superclass {self.inherits} "
1905 "is not mapped to a Table"
1906 )
1907
1908 for col in declared_columns:
1909 assert inherited_table is not None
1910 if col.name in inherited_table.c:
1911 if inherited_table.c[col.name] is col:
1912 continue
1913 raise exc.ArgumentError(
1914 f"Column '{col}' on class {cls.__name__} "
1915 f"conflicts with existing column "
1916 f"'{inherited_table.c[col.name]}'. If using "
1917 f"Declarative, consider using the "
1918 "use_existing_column parameter of mapped_column() "
1919 "to resolve conflicts."
1920 )
1921 if col.primary_key:
1922 raise exc.ArgumentError(
1923 "Can't place primary key columns on an inherited "
1924 "class with no table."
1925 )
1926
1927 if TYPE_CHECKING:
1928 assert isinstance(inherited_table, Table)
1929
1930 inherited_table.append_column(col)
1931 if (
1932 inherited_persist_selectable is not None
1933 and inherited_persist_selectable is not inherited_table
1934 ):
1935 inherited_persist_selectable._refresh_for_new_column(
1936 col
1937 )
1938
1939 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None:
1940 properties = self.properties
1941
1942 if self.mapper_args_fn:
1943 mapper_args = self.mapper_args_fn()
1944 else:
1945 mapper_args = {}
1946
1947 if mapper_kw:
1948 mapper_args.update(mapper_kw)
1949
1950 if "properties" in mapper_args:
1951 properties = dict(properties)
1952 properties.update(mapper_args["properties"])
1953
1954 # make sure that column copies are used rather
1955 # than the original columns from any mixins
1956 for k in ("version_id_col", "polymorphic_on"):
1957 if k in mapper_args:
1958 v = mapper_args[k]
1959 mapper_args[k] = self.column_copies.get(v, v)
1960
1961 if "primary_key" in mapper_args:
1962 mapper_args["primary_key"] = [
1963 self.column_copies.get(v, v)
1964 for v in util.to_list(mapper_args["primary_key"])
1965 ]
1966
1967 if "inherits" in mapper_args:
1968 inherits_arg = mapper_args["inherits"]
1969 if isinstance(inherits_arg, Mapper):
1970 inherits_arg = inherits_arg.class_
1971
1972 if inherits_arg is not self.inherits:
1973 raise exc.InvalidRequestError(
1974 "mapper inherits argument given for non-inheriting "
1975 "class %s" % (mapper_args["inherits"])
1976 )
1977
1978 if self.inherits:
1979 mapper_args["inherits"] = self.inherits
1980
1981 if self.inherits and not mapper_args.get("concrete", False):
1982 # note the superclass is expected to have a Mapper assigned and
1983 # not be a deferred config, as this is called within map()
1984 inherited_mapper = class_mapper(self.inherits, False)
1985 inherited_table = inherited_mapper.local_table
1986
1987 # single or joined inheritance
1988 # exclude any cols on the inherited table which are
1989 # not mapped on the parent class, to avoid
1990 # mapping columns specific to sibling/nephew classes
1991 if "exclude_properties" not in mapper_args:
1992 mapper_args["exclude_properties"] = exclude_properties = {
1993 c.key
1994 for c in inherited_table.c
1995 if c not in inherited_mapper._columntoproperty
1996 }.union(inherited_mapper.exclude_properties or ())
1997 exclude_properties.difference_update(
1998 [c.key for c in self.declared_columns]
1999 )
2000
2001 # look through columns in the current mapper that
2002 # are keyed to a propname different than the colname
2003 # (if names were the same, we'd have popped it out above,
2004 # in which case the mapper makes this combination).
2005 # See if the superclass has a similar column property.
2006 # If so, join them together.
2007 for k, col in list(properties.items()):
2008 if not isinstance(col, expression.ColumnElement):
2009 continue
2010 if k in inherited_mapper._props:
2011 p = inherited_mapper._props[k]
2012 if isinstance(p, ColumnProperty):
2013 # note here we place the subclass column
2014 # first. See [ticket:1892] for background.
2015 properties[k] = [col] + p.columns
2016 result_mapper_args = mapper_args.copy()
2017 result_mapper_args["properties"] = properties
2018 self.mapper_args = result_mapper_args
2019
2020 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2021 self._prepare_mapper_arguments(mapper_kw)
2022 if hasattr(self.cls, "__mapper_cls__"):
2023 mapper_cls = cast(
2024 "Type[Mapper[Any]]",
2025 util.unbound_method_to_callable(
2026 self.cls.__mapper_cls__ # type: ignore[no-untyped-call]
2027 ),
2028 )
2029 else:
2030 mapper_cls = Mapper
2031
2032 return self.set_cls_attribute(
2033 "__mapper__",
2034 mapper_cls(self.cls, self.local_table, **self.mapper_args),
2035 )
2036
2037
2038class _UnmappedDataclassConfig(_ClassScanAbstractConfig):
2039 """Configurator that will produce an unmapped dataclass."""
2040
2041 __slots__ = (
2042 "clsdict_view",
2043 "collected_attributes",
2044 "collected_annotations",
2045 "allow_dataclass_fields",
2046 "dataclass_setup_arguments",
2047 "is_dataclass_prior_to_mapping",
2048 "allow_unmapped_annotations",
2049 )
2050
2051 def __init__(
2052 self,
2053 cls_: Type[_O],
2054 dict_: _ClassDict,
2055 ):
2056 super().__init__(cls_)
2057 self.clsdict_view = (
2058 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
2059 )
2060 self.dataclass_setup_arguments = getattr(
2061 self.cls, "_sa_apply_dc_transforms", None
2062 )
2063
2064 self.is_dataclass_prior_to_mapping = dataclasses.is_dataclass(cls_)
2065 self.allow_dataclass_fields = False
2066 self.allow_unmapped_annotations = True
2067 self.collected_attributes = {}
2068 self.collected_annotations = {}
2069
2070 self._scan_attributes()
2071
2072 self._setup_dataclasses_transforms(
2073 enable_descriptor_defaults=False, revert=True
2074 )
2075
2076 def _scan_attributes(self) -> None:
2077 cls = self.cls
2078
2079 clsdict_view = self.clsdict_view
2080 collected_attributes = self.collected_attributes
2081 _include_dunders = self._include_dunders
2082
2083 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
2084
2085 local_attributes_for_class = self._cls_attr_resolver(cls)
2086 for (
2087 name,
2088 obj,
2089 annotation,
2090 is_dataclass_field,
2091 ) in local_attributes_for_class():
2092 if name in _include_dunders:
2093 continue
2094 elif is_dataclass_field and (
2095 name not in clsdict_view or clsdict_view[name] is not obj
2096 ):
2097 # here, we are definitely looking at the target class
2098 # and not a superclass. this is currently a
2099 # dataclass-only path. if the name is only
2100 # a dataclass field and isn't in local cls.__dict__,
2101 # put the object there.
2102 # assert that the dataclass-enabled resolver agrees
2103 # with what we are seeing
2104
2105 assert not attribute_is_overridden(name, obj)
2106
2107 if _is_declarative_props(obj):
2108 obj = obj.fget()
2109
2110 collected_attributes[name] = obj
2111 self._collect_annotation(name, annotation, cls, False, obj)
2112 else:
2113 self._collect_annotation(name, annotation, cls, None, obj)
2114 if name in clsdict_view:
2115 collected_attributes[name] = obj
2116
2117
2118@util.preload_module("sqlalchemy.orm.decl_api")
2119def _as_dc_declaredattr(
2120 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str
2121) -> Any:
2122 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr.
2123 # we can't write it because field.metadata is immutable :( so we have
2124 # to go through extra trouble to compare these
2125 decl_api = util.preloaded.orm_decl_api
2126 obj = field_metadata[sa_dataclass_metadata_key]
2127 if callable(obj) and not isinstance(obj, decl_api.declared_attr):
2128 return decl_api.declared_attr(obj)
2129 else:
2130 return obj
2131
2132
2133class _DeferredDeclarativeConfig(_DeclarativeMapperConfig):
2134 """Configurator that extends _DeclarativeMapperConfig to add a
2135 "deferred" step, to allow extensions like AbstractConcreteBase,
2136 DeferredMapping to partially set up a mapping that is "prepared"
2137 when table metadata is ready.
2138
2139 """
2140
2141 _cls: weakref.ref[Type[Any]]
2142
2143 is_deferred = True
2144
2145 _configs: util.OrderedDict[
2146 weakref.ref[Type[Any]], _DeferredDeclarativeConfig
2147 ] = util.OrderedDict()
2148
2149 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
2150 pass
2151
2152 @property
2153 def cls(self) -> Type[Any]:
2154 return self._cls() # type: ignore[return-value]
2155
2156 @cls.setter
2157 def cls(self, class_: Type[Any]) -> None:
2158 self._cls = weakref.ref(class_, self._remove_config_cls)
2159 self._configs[self._cls] = self
2160
2161 @classmethod
2162 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None:
2163 cls._configs.pop(ref, None)
2164
2165 @classmethod
2166 def has_cls(cls, class_: Type[Any]) -> bool:
2167 # 2.6 fails on weakref if class_ is an old style class
2168 return isinstance(class_, type) and weakref.ref(class_) in cls._configs
2169
2170 @classmethod
2171 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn:
2172 if hasattr(class_, "_sa_raise_deferred_config"):
2173 class_._sa_raise_deferred_config()
2174
2175 raise orm_exc.UnmappedClassError(
2176 class_,
2177 msg=(
2178 f"Class {orm_exc._safe_cls_name(class_)} has a deferred "
2179 "mapping on it. It is not yet usable as a mapped class."
2180 ),
2181 )
2182
2183 @classmethod
2184 def config_for_cls(cls, class_: Type[Any]) -> _DeferredDeclarativeConfig:
2185 return cls._configs[weakref.ref(class_)]
2186
2187 @classmethod
2188 def classes_for_base(
2189 cls, base_cls: Type[Any], sort: bool = True
2190 ) -> List[_DeferredDeclarativeConfig]:
2191 classes_for_base = [
2192 m
2193 for m, cls_ in [(m, m.cls) for m in cls._configs.values()]
2194 if cls_ is not None and issubclass(cls_, base_cls)
2195 ]
2196
2197 if not sort:
2198 return classes_for_base
2199
2200 all_m_by_cls = {m.cls: m for m in classes_for_base}
2201
2202 tuples: List[
2203 Tuple[_DeferredDeclarativeConfig, _DeferredDeclarativeConfig]
2204 ] = []
2205 for m_cls in all_m_by_cls:
2206 tuples.extend(
2207 (all_m_by_cls[base_cls], all_m_by_cls[m_cls])
2208 for base_cls in m_cls.__bases__
2209 if base_cls in all_m_by_cls
2210 )
2211 return list(topological.sort(tuples, classes_for_base))
2212
2213 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2214 self._configs.pop(self._cls, None)
2215 return super().map(mapper_kw)
2216
2217
2218def _add_attribute(
2219 cls: Type[Any], key: str, value: MapperProperty[Any]
2220) -> None:
2221 """add an attribute to an existing declarative class.
2222
2223 This runs through the logic to determine MapperProperty,
2224 adds it to the Mapper, adds a column to the mapped Table, etc.
2225
2226 """
2227
2228 if "__mapper__" in cls.__dict__:
2229 mapped_cls = cast("MappedClassProtocol[Any]", cls)
2230
2231 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table:
2232 if isinstance(mc.__table__, Table):
2233 return mc.__table__
2234 raise exc.InvalidRequestError(
2235 f"Cannot add a new attribute to mapped class {mc.__name__!r} "
2236 "because it's not mapped against a table."
2237 )
2238
2239 if isinstance(value, Column):
2240 _undefer_column_name(key, value)
2241 _table_or_raise(mapped_cls).append_column(
2242 value, replace_existing=True
2243 )
2244 mapped_cls.__mapper__.add_property(key, value)
2245 elif isinstance(value, _MapsColumns):
2246 mp = value.mapper_property_to_assign
2247 for col, _ in value.columns_to_assign:
2248 _undefer_column_name(key, col)
2249 _table_or_raise(mapped_cls).append_column(
2250 col, replace_existing=True
2251 )
2252 if not mp:
2253 mapped_cls.__mapper__.add_property(key, col)
2254 if mp:
2255 mapped_cls.__mapper__.add_property(key, mp)
2256 elif isinstance(value, MapperProperty):
2257 mapped_cls.__mapper__.add_property(key, value)
2258 elif isinstance(value, QueryableAttribute) and value.key != key:
2259 # detect a QueryableAttribute that's already mapped being
2260 # assigned elsewhere in userland, turn into a synonym()
2261 value = SynonymProperty(value.key)
2262 mapped_cls.__mapper__.add_property(key, value)
2263 else:
2264 type.__setattr__(cls, key, value)
2265 mapped_cls.__mapper__._expire_memoizations()
2266 else:
2267 type.__setattr__(cls, key, value)
2268
2269
2270def _del_attribute(cls: Type[Any], key: str) -> None:
2271 if (
2272 "__mapper__" in cls.__dict__
2273 and key in cls.__dict__
2274 and not cast(
2275 "MappedClassProtocol[Any]", cls
2276 ).__mapper__._dispose_called
2277 ):
2278 value = cls.__dict__[key]
2279 if isinstance(
2280 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute)
2281 ):
2282 raise NotImplementedError(
2283 "Can't un-map individual mapped attributes on a mapped class."
2284 )
2285 else:
2286 type.__delattr__(cls, key)
2287 cast(
2288 "MappedClassProtocol[Any]", cls
2289 ).__mapper__._expire_memoizations()
2290 else:
2291 type.__delattr__(cls, key)
2292
2293
2294def _declarative_constructor(self: Any, **kwargs: Any) -> None:
2295 """A simple constructor that allows initialization from kwargs.
2296
2297 Sets attributes on the constructed instance using the names and
2298 values in ``kwargs``.
2299
2300 Only keys that are present as
2301 attributes of the instance's class are allowed. These could be,
2302 for example, any mapped columns or relationships.
2303 """
2304 cls_ = type(self)
2305 for k in kwargs:
2306 if not hasattr(cls_, k):
2307 raise TypeError(
2308 "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
2309 )
2310 setattr(self, k, kwargs[k])
2311
2312
2313_declarative_constructor.__name__ = "__init__"
2314
2315
2316def _undefer_column_name(key: str, column: Column[Any]) -> None:
2317 if column.key is None:
2318 column.key = key
2319 if column.name is None:
2320 column.name = key