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