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 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
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
87_T = TypeVar("_T", bound=Any)
89_MapperKwArgs = Mapping[str, Any]
90_TableArgsType = Union[Tuple[Any, ...], Dict[str, Any]]
93class MappedClassProtocol(Protocol[_O]):
94 """A protocol representing a SQLAlchemy mapped class.
96 The protocol is generic on the type of class, use
97 ``MappedClassProtocol[Any]`` to allow any mapped class.
98 """
100 __name__: str
101 __mapper__: Mapper[_O]
102 __table__: FromClause
104 def __call__(self, **kw: Any) -> _O: ...
107class _DeclMappedClassProtocol(MappedClassProtocol[_O], Protocol):
108 "Internal more detailed version of ``MappedClassProtocol``."
110 metadata: MetaData
111 __tablename__: str
112 __mapper_args__: _MapperKwArgs
113 __table_args__: Optional[_TableArgsType]
115 _sa_apply_dc_transforms: Optional[_DataclassArguments]
117 def __declare_first__(self) -> None: ...
119 def __declare_last__(self) -> None: ...
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
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'.
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()
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
157def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
158 if cls is object:
159 return None
161 sup: Optional[Type[Any]]
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)
173 if clsmanager:
174 return clsmanager.class_
175 else:
176 return cls
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.
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.
191 """
193 # the rules are different for this name than others,
194 # make sure we've moved it out. transitional
195 assert attrname != "__abstract__"
197 if not issubclass(cls, object):
198 return None
200 if attrname in cls.__dict__:
201 return getattr(cls, attrname)
203 for base in cls.__mro__[1:]:
204 _is_classical_inherits = _dive_for_cls_manager(base) is not None
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
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"
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
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
236 return isinstance(obj, (_declared_attr_common, util.classproperty))
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
254class _ORMClassConfigurator:
255 """Object that configures a class that's potentially going to be
256 mapped, and/or turned into an ORM dataclass.
258 This is the base class for all the configurator objects.
260 """
262 __slots__ = ("cls", "classname", "__weakref__")
264 cls: Type[Any]
265 classname: str
267 def __init__(self, cls_: Type[Any]):
268 self.cls = util.assert_arg_type(cls_, type, "cls_")
269 self.classname = cls_.__name__
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 )
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
288 defer_map = _get_immediate_cls_attr(
289 cls_, "_sa_decl_prepare_nocascade", strict=True
290 ) or hasattr(cls_, "_sa_decl_prepare")
292 if defer_map:
293 return _DeferredDeclarativeConfig(registry, cls_, dict_)
294 else:
295 return _DeclarativeMapperConfig(registry, cls_, dict_)
297 @classmethod
298 def _as_unmapped_dataclass(
299 cls, cls_: Type[Any], dict_: _ClassDict
300 ) -> _UnmappedDataclassConfig:
301 return _UnmappedDataclassConfig(cls_, dict_)
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__
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."""
319 __slots__ = (
320 "properties",
321 "declared_attr_reg",
322 )
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]
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 = {}
341 instrumentation.register_class(
342 self.cls,
343 finalize=False,
344 registry=registry,
345 declarative_scan=self,
346 init_method=registry.constructor,
347 )
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
354 def map(self, mapper_kw: _MapperKwArgs) -> Mapper[Any]:
355 raise NotImplementedError()
357 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
358 self.map(mapper_kw)
361class _ImperativeMapperConfig(_MapperConfig):
362 """Configurator that configures a class for an imperative mapping."""
364 __slots__ = ("local_table", "inherits")
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_)
375 self.local_table = self.set_cls_attribute("__table__", table)
377 with mapperlib._CONFIGURE_MUTEX:
378 clsregistry._add_class(
379 self.classname, self.cls, registry._class_registry
380 )
382 self._setup_inheritance(mapper_kw)
384 self._early_mapping(mapper_kw)
386 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
387 mapper_cls = Mapper
389 return self.set_cls_attribute(
390 "__mapper__",
391 mapper_cls(self.cls, self.local_table, **mapper_kw),
392 )
394 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
395 cls = self.cls
397 inherits = None
398 inherits_search = []
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
407 if _is_supercls_for_inherits(c) and c not in inherits_search:
408 inherits_search.append(c)
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]
418 self.inherits = inherits
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]
431class _ClassScanAbstractConfig(_ORMClassConfigurator):
432 """Abstract base for a configurator that configures a class for a
433 declarative mapping, or an unmapped ORM dataclass.
435 Defines scanning of pep-484 annotations as well as ORM dataclass
436 applicators
438 """
440 __slots__ = ()
442 clsdict_view: _ClassDict
443 collected_annotations: Dict[str, _CollectedAnnotation]
444 collected_attributes: Dict[str, Any]
446 is_dataclass_prior_to_mapping: bool
447 allow_unmapped_annotations: bool
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.
454 """
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.
461 if False, dataclass fields can still be used, however they won't be
462 mapped.
464 """
466 _include_dunders = {
467 "__table__",
468 "__mapper_args__",
469 "__tablename__",
470 "__table_args__",
471 }
473 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
475 def _scan_attributes(self) -> None:
476 raise NotImplementedError()
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
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 )
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 )
506 raise_for_non_dc_attrs = collections.defaultdict(list)
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)
517 return True
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 )
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
591 # get original annotations using ForwardRef for symbols that
592 # are unresolvable
593 orig_annotations = util.get_annotations(self.cls)
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 = {}
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
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)
615 for k, v in defaults.items():
616 setattr(self.cls, k, v)
618 self._assert_dc_arguments(dataclass_setup_arguments)
620 dataclass_callable = dataclass_setup_arguments["dataclass_callable"]
621 if dataclass_callable is _NoArg.NO_ARG:
622 dataclass_callable = dataclasses.dataclass
624 # create a merged __annotations__ dictionary, maintaining order
625 # as best we can:
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 }
637 # 2. then put in all the dataclass annotations we have
638 new_annotations |= swap_annotations
640 # 3. them merge all of orig_annotations which will add remaining
641 # keys
642 new_annotations |= orig_annotations
644 # 4. this becomes the new class annotations.
645 restore_anno = util.restore_annotations(self.cls, new_annotations)
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 finally:
666 if revert and revert_dict:
667 # used for mixin dataclasses; we have to restore the
668 # mapped_column(), relationship() etc. to the class so these
669 # take place for a mapped class scan
670 for k, v in revert_dict.items():
671 setattr(self.cls, k, v)
673 restore_anno()
675 def _collect_annotation(
676 self,
677 name: str,
678 raw_annotation: _AnnotationScanType,
679 originating_class: Type[Any],
680 expect_mapped: Optional[bool],
681 attr_value: Any,
682 ) -> Optional[_CollectedAnnotation]:
683 if name in self.collected_annotations:
684 return self.collected_annotations[name]
686 if raw_annotation is None:
687 return None
689 is_dataclass = self.is_dataclass_prior_to_mapping
690 allow_unmapped = self.allow_unmapped_annotations
692 if expect_mapped is None:
693 is_dataclass_field = isinstance(attr_value, dataclasses.Field)
694 expect_mapped = (
695 not is_dataclass_field
696 and not allow_unmapped
697 and (
698 attr_value is None
699 or isinstance(attr_value, _MappedAttribute)
700 )
701 )
703 is_dataclass_field = False
704 extracted = _extract_mapped_subtype(
705 raw_annotation,
706 self.cls,
707 originating_class.__module__,
708 name,
709 type(attr_value),
710 required=False,
711 is_dataclass_field=is_dataclass_field,
712 expect_mapped=expect_mapped and not is_dataclass,
713 )
714 if extracted is None:
715 # ClassVar can come out here
716 return None
718 extracted_mapped_annotation, mapped_container = extracted
720 if attr_value is None and not is_literal(extracted_mapped_annotation):
721 for elem in get_args(extracted_mapped_annotation):
722 if is_fwd_ref(
723 elem, check_generic=True, check_for_plain_string=True
724 ):
725 elem = de_stringify_annotation(
726 self.cls,
727 elem,
728 originating_class.__module__,
729 include_generic=True,
730 )
731 # look in Annotated[...] for an ORM construct,
732 # such as Annotated[int, mapped_column(primary_key=True)]
733 if isinstance(elem, _IntrospectsAnnotations):
734 attr_value = elem.found_in_pep593_annotated()
736 self.collected_annotations[name] = ca = _CollectedAnnotation(
737 raw_annotation,
738 mapped_container,
739 extracted_mapped_annotation,
740 is_dataclass,
741 attr_value,
742 originating_class.__module__,
743 originating_class,
744 )
745 return ca
747 @classmethod
748 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None:
749 allowed = {
750 "init",
751 "repr",
752 "order",
753 "eq",
754 "unsafe_hash",
755 "kw_only",
756 "match_args",
757 "dataclass_callable",
758 }
759 disallowed_args = set(arguments).difference(allowed)
760 if disallowed_args:
761 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args))
762 raise exc.ArgumentError(
763 f"Dataclass argument(s) {msg} are not accepted"
764 )
766 def _cls_attr_override_checker(
767 self, cls: Type[_O]
768 ) -> Callable[[str, Any], bool]:
769 """Produce a function that checks if a class has overridden an
770 attribute, taking SQLAlchemy-enabled dataclass fields into account.
772 """
774 if self.allow_dataclass_fields:
775 sa_dataclass_metadata_key = _get_immediate_cls_attr(
776 cls, "__sa_dataclass_metadata_key__"
777 )
778 else:
779 sa_dataclass_metadata_key = None
781 if not sa_dataclass_metadata_key:
783 def attribute_is_overridden(key: str, obj: Any) -> bool:
784 return getattr(cls, key, obj) is not obj
786 else:
787 all_datacls_fields = {
788 f.name: f.metadata[sa_dataclass_metadata_key]
789 for f in util.dataclass_fields(cls)
790 if sa_dataclass_metadata_key in f.metadata
791 }
792 local_datacls_fields = {
793 f.name: f.metadata[sa_dataclass_metadata_key]
794 for f in util.local_dataclass_fields(cls)
795 if sa_dataclass_metadata_key in f.metadata
796 }
798 absent = object()
800 def attribute_is_overridden(key: str, obj: Any) -> bool:
801 if _is_declarative_props(obj):
802 obj = obj.fget
804 # this function likely has some failure modes still if
805 # someone is doing a deep mixing of the same attribute
806 # name as plain Python attribute vs. dataclass field.
808 ret = local_datacls_fields.get(key, absent)
809 if _is_declarative_props(ret):
810 ret = ret.fget
812 if ret is obj:
813 return False
814 elif ret is not absent:
815 return True
817 all_field = all_datacls_fields.get(key, absent)
819 ret = getattr(cls, key, obj)
821 if ret is obj:
822 return False
824 # for dataclasses, this could be the
825 # 'default' of the field. so filter more specifically
826 # for an already-mapped InstrumentedAttribute
827 if ret is not absent and isinstance(
828 ret, InstrumentedAttribute
829 ):
830 return True
832 if all_field is obj:
833 return False
834 elif all_field is not absent:
835 return True
837 # can't find another attribute
838 return False
840 return attribute_is_overridden
842 def _cls_attr_resolver(
843 self, cls: Type[Any]
844 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
845 """produce a function to iterate the "attributes" of a class
846 which we want to consider for mapping, adjusting for SQLAlchemy fields
847 embedded in dataclass fields.
849 """
850 cls_annotations = util.get_annotations(cls)
852 cls_vars = vars(cls)
854 _include_dunders = self._include_dunders
855 _match_exclude_dunders = self._match_exclude_dunders
857 names = [
858 n
859 for n in util.merge_lists_w_ordering(
860 list(cls_vars), list(cls_annotations)
861 )
862 if not _match_exclude_dunders.match(n) or n in _include_dunders
863 ]
865 if self.allow_dataclass_fields:
866 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
867 cls, "__sa_dataclass_metadata_key__"
868 )
869 else:
870 sa_dataclass_metadata_key = None
872 if not sa_dataclass_metadata_key:
874 def local_attributes_for_class() -> (
875 Iterable[Tuple[str, Any, Any, bool]]
876 ):
877 return (
878 (
879 name,
880 cls_vars.get(name),
881 cls_annotations.get(name),
882 False,
883 )
884 for name in names
885 )
887 else:
888 dataclass_fields = {
889 field.name: field for field in util.local_dataclass_fields(cls)
890 }
892 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
894 def local_attributes_for_class() -> (
895 Iterable[Tuple[str, Any, Any, bool]]
896 ):
897 for name in names:
898 field = dataclass_fields.get(name, None)
899 if field and sa_dataclass_metadata_key in field.metadata:
900 yield field.name, _as_dc_declaredattr(
901 field.metadata, fixed_sa_dataclass_metadata_key
902 ), cls_annotations.get(field.name), True
903 else:
904 yield name, cls_vars.get(name), cls_annotations.get(
905 name
906 ), False
908 return local_attributes_for_class
911class _DeclarativeMapperConfig(_MapperConfig, _ClassScanAbstractConfig):
912 """Configurator that will produce a declarative mapped class"""
914 __slots__ = (
915 "registry",
916 "local_table",
917 "persist_selectable",
918 "declared_columns",
919 "column_ordering",
920 "column_copies",
921 "table_args",
922 "tablename",
923 "mapper_args",
924 "mapper_args_fn",
925 "table_fn",
926 "inherits",
927 "single",
928 "clsdict_view",
929 "collected_attributes",
930 "collected_annotations",
931 "allow_dataclass_fields",
932 "dataclass_setup_arguments",
933 "is_dataclass_prior_to_mapping",
934 "allow_unmapped_annotations",
935 )
937 is_deferred = False
938 registry: _RegistryType
939 local_table: Optional[FromClause]
940 persist_selectable: Optional[FromClause]
941 declared_columns: util.OrderedSet[Column[Any]]
942 column_ordering: Dict[Column[Any], int]
943 column_copies: Dict[
944 Union[MappedColumn[Any], Column[Any]],
945 Union[MappedColumn[Any], Column[Any]],
946 ]
947 tablename: Optional[str]
948 mapper_args: Mapping[str, Any]
949 table_args: Optional[_TableArgsType]
950 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
951 inherits: Optional[Type[Any]]
952 single: bool
954 def __init__(
955 self,
956 registry: _RegistryType,
957 cls_: Type[_O],
958 dict_: _ClassDict,
959 ):
960 # grab class dict before the instrumentation manager has been added.
961 # reduces cycles
962 self.clsdict_view = (
963 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
964 )
965 super().__init__(registry, cls_)
966 self.registry = registry
967 self.persist_selectable = None
969 self.collected_attributes = {}
970 self.collected_annotations = {}
971 self.declared_columns = util.OrderedSet()
972 self.column_ordering = {}
973 self.column_copies = {}
974 self.single = False
975 self.dataclass_setup_arguments = dca = getattr(
976 self.cls, "_sa_apply_dc_transforms", None
977 )
979 self.allow_unmapped_annotations = getattr(
980 self.cls, "__allow_unmapped__", False
981 ) or bool(self.dataclass_setup_arguments)
983 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
984 cls_
985 )
987 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
989 # we don't want to consume Field objects from a not-already-dataclass.
990 # the Field objects won't have their "name" or "type" populated,
991 # and while it seems like we could just set these on Field as we
992 # read them, Field is documented as "user read only" and we need to
993 # stay far away from any off-label use of dataclasses APIs.
994 if (not cld or dca) and sdk:
995 raise exc.InvalidRequestError(
996 "SQLAlchemy mapped dataclasses can't consume mapping "
997 "information from dataclass.Field() objects if the immediate "
998 "class is not already a dataclass."
999 )
1001 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
1002 # then also look inside of dataclass.Field() objects yielded by
1003 # dataclasses.get_fields(cls) when scanning for attributes
1004 self.allow_dataclass_fields = bool(sdk and cld)
1006 self._setup_declared_events()
1008 self._scan_attributes()
1010 self._setup_dataclasses_transforms(enable_descriptor_defaults=True)
1012 with mapperlib._CONFIGURE_MUTEX:
1013 clsregistry._add_class(
1014 self.classname, self.cls, registry._class_registry
1015 )
1017 self._setup_inheriting_mapper()
1019 self._extract_mappable_attributes()
1021 self._extract_declared_columns()
1023 self._setup_table()
1025 self._setup_inheriting_columns()
1027 self._early_mapping(util.EMPTY_DICT)
1029 def _setup_declared_events(self) -> None:
1030 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
1032 @event.listens_for(Mapper, "after_configured")
1033 def after_configured() -> None:
1034 cast(
1035 "_DeclMappedClassProtocol[Any]", self.cls
1036 ).__declare_last__()
1038 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
1040 @event.listens_for(Mapper, "before_configured")
1041 def before_configured() -> None:
1042 cast(
1043 "_DeclMappedClassProtocol[Any]", self.cls
1044 ).__declare_first__()
1046 def _scan_attributes(self) -> None:
1047 cls = self.cls
1049 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
1051 clsdict_view = self.clsdict_view
1052 collected_attributes = self.collected_attributes
1053 column_copies = self.column_copies
1054 _include_dunders = self._include_dunders
1055 mapper_args_fn = None
1056 table_args = inherited_table_args = None
1057 table_fn = None
1058 tablename = None
1059 fixed_table = "__table__" in clsdict_view
1061 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
1063 bases = []
1065 for base in cls.__mro__:
1066 # collect bases and make sure standalone columns are copied
1067 # to be the column they will ultimately be on the class,
1068 # so that declared_attr functions use the right columns.
1069 # need to do this all the way up the hierarchy first
1070 # (see #8190)
1072 class_mapped = base is not cls and _is_supercls_for_inherits(base)
1074 local_attributes_for_class = self._cls_attr_resolver(base)
1076 if not class_mapped and base is not cls:
1077 locally_collected_columns = self._produce_column_copies(
1078 local_attributes_for_class,
1079 attribute_is_overridden,
1080 fixed_table,
1081 base,
1082 )
1083 else:
1084 locally_collected_columns = {}
1086 bases.append(
1087 (
1088 base,
1089 class_mapped,
1090 local_attributes_for_class,
1091 locally_collected_columns,
1092 )
1093 )
1095 for (
1096 base,
1097 class_mapped,
1098 local_attributes_for_class,
1099 locally_collected_columns,
1100 ) in bases:
1101 # this transfer can also take place as we scan each name
1102 # for finer-grained control of how collected_attributes is
1103 # populated, as this is what impacts column ordering.
1104 # however it's simpler to get it out of the way here.
1105 collected_attributes.update(locally_collected_columns)
1107 for (
1108 name,
1109 obj,
1110 annotation,
1111 is_dataclass_field,
1112 ) in local_attributes_for_class():
1113 if name in _include_dunders:
1114 if name == "__mapper_args__":
1115 check_decl = _check_declared_props_nocascade(
1116 obj, name, cls
1117 )
1118 if not mapper_args_fn and (
1119 not class_mapped or check_decl
1120 ):
1121 # don't even invoke __mapper_args__ until
1122 # after we've determined everything about the
1123 # mapped table.
1124 # make a copy of it so a class-level dictionary
1125 # is not overwritten when we update column-based
1126 # arguments.
1127 def _mapper_args_fn() -> Dict[str, Any]:
1128 return dict(cls_as_Decl.__mapper_args__)
1130 mapper_args_fn = _mapper_args_fn
1132 elif name == "__tablename__":
1133 check_decl = _check_declared_props_nocascade(
1134 obj, name, cls
1135 )
1136 if not tablename and (not class_mapped or check_decl):
1137 tablename = cls_as_Decl.__tablename__
1138 elif name == "__table__":
1139 check_decl = _check_declared_props_nocascade(
1140 obj, name, cls
1141 )
1142 # if a @declared_attr using "__table__" is detected,
1143 # wrap up a callable to look for "__table__" from
1144 # the final concrete class when we set up a table.
1145 # this was fixed by
1146 # #11509, regression in 2.0 from version 1.4.
1147 if check_decl and not table_fn:
1148 # don't even invoke __table__ until we're ready
1149 def _table_fn() -> FromClause:
1150 return cls_as_Decl.__table__
1152 table_fn = _table_fn
1154 elif name == "__table_args__":
1155 check_decl = _check_declared_props_nocascade(
1156 obj, name, cls
1157 )
1158 if not table_args and (not class_mapped or check_decl):
1159 table_args = cls_as_Decl.__table_args__
1160 if not isinstance(
1161 table_args, (tuple, dict, type(None))
1162 ):
1163 raise exc.ArgumentError(
1164 "__table_args__ value must be a tuple, "
1165 "dict, or None"
1166 )
1167 if base is not cls:
1168 inherited_table_args = True
1169 else:
1170 # any other dunder names; should not be here
1171 # as we have tested for all four names in
1172 # _include_dunders
1173 assert False
1174 elif class_mapped:
1175 if _is_declarative_props(obj) and not obj._quiet:
1176 util.warn(
1177 "Regular (i.e. not __special__) "
1178 "attribute '%s.%s' uses @declared_attr, "
1179 "but owning class %s is mapped - "
1180 "not applying to subclass %s."
1181 % (base.__name__, name, base, cls)
1182 )
1184 continue
1185 elif base is not cls:
1186 # we're a mixin, abstract base, or something that is
1187 # acting like that for now.
1189 if isinstance(obj, (Column, MappedColumn)):
1190 # already copied columns to the mapped class.
1191 continue
1192 elif isinstance(obj, MapperProperty):
1193 raise exc.InvalidRequestError(
1194 "Mapper properties (i.e. deferred,"
1195 "column_property(), relationship(), etc.) must "
1196 "be declared as @declared_attr callables "
1197 "on declarative mixin classes. For dataclass "
1198 "field() objects, use a lambda:"
1199 )
1200 elif _is_declarative_props(obj):
1201 # tried to get overloads to tell this to
1202 # pylance, no luck
1203 assert obj is not None
1205 if obj._cascading:
1206 if name in clsdict_view:
1207 # unfortunately, while we can use the user-
1208 # defined attribute here to allow a clean
1209 # override, if there's another
1210 # subclass below then it still tries to use
1211 # this. not sure if there is enough
1212 # information here to add this as a feature
1213 # later on.
1214 util.warn(
1215 "Attribute '%s' on class %s cannot be "
1216 "processed due to "
1217 "@declared_attr.cascading; "
1218 "skipping" % (name, cls)
1219 )
1220 collected_attributes[name] = column_copies[obj] = (
1221 ret
1222 ) = obj.__get__(obj, cls)
1223 setattr(cls, name, ret)
1224 else:
1225 if is_dataclass_field:
1226 # access attribute using normal class access
1227 # first, to see if it's been mapped on a
1228 # superclass. note if the dataclasses.field()
1229 # has "default", this value can be anything.
1230 ret = getattr(cls, name, None)
1232 # so, if it's anything that's not ORM
1233 # mapped, assume we should invoke the
1234 # declared_attr
1235 if not isinstance(ret, InspectionAttr):
1236 ret = obj.fget()
1237 else:
1238 # access attribute using normal class access.
1239 # if the declared attr already took place
1240 # on a superclass that is mapped, then
1241 # this is no longer a declared_attr, it will
1242 # be the InstrumentedAttribute
1243 ret = getattr(cls, name)
1245 # correct for proxies created from hybrid_property
1246 # or similar. note there is no known case that
1247 # produces nested proxies, so we are only
1248 # looking one level deep right now.
1250 if (
1251 isinstance(ret, InspectionAttr)
1252 and attr_is_internal_proxy(ret)
1253 and not isinstance(
1254 ret.original_property, MapperProperty
1255 )
1256 ):
1257 ret = ret.descriptor
1259 collected_attributes[name] = column_copies[obj] = (
1260 ret
1261 )
1263 if (
1264 isinstance(ret, (Column, MapperProperty))
1265 and ret.doc is None
1266 ):
1267 ret.doc = obj.__doc__
1269 self._collect_annotation(
1270 name,
1271 obj._collect_return_annotation(),
1272 base,
1273 True,
1274 obj,
1275 )
1276 elif _is_mapped_annotation(annotation, cls, base):
1277 # Mapped annotation without any object.
1278 # product_column_copies should have handled this.
1279 # if future support for other MapperProperty,
1280 # then test if this name is already handled and
1281 # otherwise proceed to generate.
1282 if not fixed_table:
1283 assert (
1284 name in collected_attributes
1285 or attribute_is_overridden(name, None)
1286 )
1287 continue
1288 else:
1289 # here, the attribute is some other kind of
1290 # property that we assume is not part of the
1291 # declarative mapping. however, check for some
1292 # more common mistakes
1293 self._warn_for_decl_attributes(base, name, obj)
1294 elif is_dataclass_field and (
1295 name not in clsdict_view or clsdict_view[name] is not obj
1296 ):
1297 # here, we are definitely looking at the target class
1298 # and not a superclass. this is currently a
1299 # dataclass-only path. if the name is only
1300 # a dataclass field and isn't in local cls.__dict__,
1301 # put the object there.
1302 # assert that the dataclass-enabled resolver agrees
1303 # with what we are seeing
1305 assert not attribute_is_overridden(name, obj)
1307 if _is_declarative_props(obj):
1308 obj = obj.fget()
1310 collected_attributes[name] = obj
1311 self._collect_annotation(
1312 name, annotation, base, False, obj
1313 )
1314 else:
1315 collected_annotation = self._collect_annotation(
1316 name, annotation, base, None, obj
1317 )
1318 is_mapped = (
1319 collected_annotation is not None
1320 and collected_annotation.mapped_container is not None
1321 )
1322 generated_obj = (
1323 collected_annotation.attr_value
1324 if collected_annotation is not None
1325 else obj
1326 )
1327 if obj is None and not fixed_table and is_mapped:
1328 collected_attributes[name] = (
1329 generated_obj
1330 if generated_obj is not None
1331 else MappedColumn()
1332 )
1333 elif name in clsdict_view:
1334 collected_attributes[name] = obj
1335 # else if the name is not in the cls.__dict__,
1336 # don't collect it as an attribute.
1337 # we will see the annotation only, which is meaningful
1338 # both for mapping and dataclasses setup
1340 if inherited_table_args and not tablename:
1341 table_args = None
1343 self.table_args = table_args
1344 self.tablename = tablename
1345 self.mapper_args_fn = mapper_args_fn
1346 self.table_fn = table_fn
1348 @classmethod
1349 def _update_annotations_for_non_mapped_class(
1350 cls, klass: Type[_O]
1351 ) -> Mapping[str, _AnnotationScanType]:
1352 cls_annotations = util.get_annotations(klass)
1354 new_anno = {}
1355 for name, annotation in cls_annotations.items():
1356 if _is_mapped_annotation(annotation, klass, klass):
1357 extracted = _extract_mapped_subtype(
1358 annotation,
1359 klass,
1360 klass.__module__,
1361 name,
1362 type(None),
1363 required=False,
1364 is_dataclass_field=False,
1365 expect_mapped=False,
1366 )
1367 if extracted:
1368 inner, _ = extracted
1369 new_anno[name] = inner
1370 else:
1371 new_anno[name] = annotation
1372 return new_anno
1374 def _warn_for_decl_attributes(
1375 self, cls: Type[Any], key: str, c: Any
1376 ) -> None:
1377 if isinstance(c, expression.ColumnElement):
1378 util.warn(
1379 f"Attribute '{key}' on class {cls} appears to "
1380 "be a non-schema SQLAlchemy expression "
1381 "object; this won't be part of the declarative mapping. "
1382 "To map arbitrary expressions, use ``column_property()`` "
1383 "or a similar function such as ``deferred()``, "
1384 "``query_expression()`` etc. "
1385 )
1387 def _produce_column_copies(
1388 self,
1389 attributes_for_class: Callable[
1390 [], Iterable[Tuple[str, Any, Any, bool]]
1391 ],
1392 attribute_is_overridden: Callable[[str, Any], bool],
1393 fixed_table: bool,
1394 originating_class: Type[Any],
1395 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]:
1396 cls = self.cls
1397 dict_ = self.clsdict_view
1398 locally_collected_attributes = {}
1399 column_copies = self.column_copies
1400 # copy mixin columns to the mapped class
1402 for name, obj, annotation, is_dataclass in attributes_for_class():
1403 if (
1404 not fixed_table
1405 and obj is None
1406 and _is_mapped_annotation(annotation, cls, originating_class)
1407 ):
1408 # obj is None means this is the annotation only path
1410 if attribute_is_overridden(name, obj):
1411 # perform same "overridden" check as we do for
1412 # Column/MappedColumn, this is how a mixin col is not
1413 # applied to an inherited subclass that does not have
1414 # the mixin. the anno-only path added here for
1415 # #9564
1416 continue
1418 collected_annotation = self._collect_annotation(
1419 name, annotation, originating_class, True, obj
1420 )
1421 obj = (
1422 collected_annotation.attr_value
1423 if collected_annotation is not None
1424 else obj
1425 )
1426 if obj is None:
1427 obj = MappedColumn()
1429 locally_collected_attributes[name] = obj
1430 setattr(cls, name, obj)
1432 elif isinstance(obj, (Column, MappedColumn)):
1433 if attribute_is_overridden(name, obj):
1434 # if column has been overridden
1435 # (like by the InstrumentedAttribute of the
1436 # superclass), skip. don't collect the annotation
1437 # either (issue #8718)
1438 continue
1440 collected_annotation = self._collect_annotation(
1441 name, annotation, originating_class, True, obj
1442 )
1443 obj = (
1444 collected_annotation.attr_value
1445 if collected_annotation is not None
1446 else obj
1447 )
1449 if name not in dict_ and not (
1450 "__table__" in dict_
1451 and (getattr(obj, "name", None) or name)
1452 in dict_["__table__"].c
1453 ):
1454 if obj.foreign_keys:
1455 for fk in obj.foreign_keys:
1456 if (
1457 fk._table_column is not None
1458 and fk._table_column.table is None
1459 ):
1460 raise exc.InvalidRequestError(
1461 "Columns with foreign keys to "
1462 "non-table-bound "
1463 "columns must be declared as "
1464 "@declared_attr callables "
1465 "on declarative mixin classes. "
1466 "For dataclass "
1467 "field() objects, use a lambda:."
1468 )
1470 column_copies[obj] = copy_ = obj._copy()
1472 locally_collected_attributes[name] = copy_
1473 setattr(cls, name, copy_)
1475 return locally_collected_attributes
1477 def _extract_mappable_attributes(self) -> None:
1478 cls = self.cls
1479 collected_attributes = self.collected_attributes
1481 our_stuff = self.properties
1483 _include_dunders = self._include_dunders
1485 late_mapped = _get_immediate_cls_attr(
1486 cls, "_sa_decl_prepare_nocascade", strict=True
1487 )
1489 allow_unmapped_annotations = self.allow_unmapped_annotations
1490 expect_annotations_wo_mapped = (
1491 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping
1492 )
1494 look_for_dataclass_things = bool(self.dataclass_setup_arguments)
1496 for k in list(collected_attributes):
1497 if k in _include_dunders:
1498 continue
1500 value = collected_attributes[k]
1502 if _is_declarative_props(value):
1503 # @declared_attr in collected_attributes only occurs here for a
1504 # @declared_attr that's directly on the mapped class;
1505 # for a mixin, these have already been evaluated
1506 if value._cascading:
1507 util.warn(
1508 "Use of @declared_attr.cascading only applies to "
1509 "Declarative 'mixin' and 'abstract' classes. "
1510 "Currently, this flag is ignored on mapped class "
1511 "%s" % self.cls
1512 )
1514 value = getattr(cls, k)
1516 elif (
1517 isinstance(value, QueryableAttribute)
1518 and value.class_ is not cls
1519 and value.key != k
1520 ):
1521 # detect a QueryableAttribute that's already mapped being
1522 # assigned elsewhere in userland, turn into a synonym()
1523 value = SynonymProperty(value.key)
1524 setattr(cls, k, value)
1526 if k in ("metadata", "registry"):
1527 noun = {
1528 "metadata": "MetaData instance",
1529 "registry": "Declarative class registry",
1530 }
1531 util.warn(
1532 f"Attribute name '{k}' should be left reserved for the "
1533 f"{noun[k]} when using a Declarative class."
1534 )
1536 if (
1537 isinstance(value, tuple)
1538 and len(value) == 1
1539 and isinstance(value[0], (Column, _MappedAttribute))
1540 ):
1541 util.warn(
1542 "Ignoring declarative-like tuple value of attribute "
1543 "'%s': possibly a copy-and-paste error with a comma "
1544 "accidentally placed at the end of the line?" % k
1545 )
1546 continue
1547 elif look_for_dataclass_things and isinstance(
1548 value, dataclasses.Field
1549 ):
1550 # we collected a dataclass Field; dataclasses would have
1551 # set up the correct state on the class
1552 continue
1553 elif not isinstance(value, (Column, _DCAttributeOptions)):
1554 # using @declared_attr for some object that
1555 # isn't Column/MapperProperty/_DCAttributeOptions; remove
1556 # from the clsdict_view
1557 # and place the evaluated value onto the class.
1558 collected_attributes.pop(k)
1559 self._warn_for_decl_attributes(cls, k, value)
1560 if not late_mapped:
1561 setattr(cls, k, value)
1562 continue
1563 elif isinstance(value, Column):
1564 _undefer_column_name(
1565 k, self.column_copies.get(value, value) # type: ignore
1566 )
1567 else:
1568 if isinstance(value, _IntrospectsAnnotations):
1569 (
1570 annotation,
1571 mapped_container,
1572 extracted_mapped_annotation,
1573 is_dataclass,
1574 attr_value,
1575 originating_module,
1576 originating_class,
1577 ) = self.collected_annotations.get(
1578 k, (None, None, None, False, None, None, None)
1579 )
1581 # issue #8692 - don't do any annotation interpretation if
1582 # an annotation were present and a container such as
1583 # Mapped[] etc. were not used. If annotation is None,
1584 # do declarative_scan so that the property can raise
1585 # for required
1586 if (
1587 mapped_container is not None
1588 or annotation is None
1589 # issue #10516: need to do declarative_scan even with
1590 # a non-Mapped annotation if we are doing
1591 # __allow_unmapped__, for things like col.name
1592 # assignment
1593 or allow_unmapped_annotations
1594 ):
1595 try:
1596 value.declarative_scan(
1597 self,
1598 self.registry,
1599 cls,
1600 originating_module,
1601 k,
1602 mapped_container,
1603 annotation,
1604 extracted_mapped_annotation,
1605 is_dataclass,
1606 )
1607 except NameError as ne:
1608 raise orm_exc.MappedAnnotationError(
1609 f"Could not resolve all types within mapped "
1610 f'annotation: "{annotation}". Ensure all '
1611 f"types are written correctly and are "
1612 f"imported within the module in use."
1613 ) from ne
1614 else:
1615 # assert that we were expecting annotations
1616 # without Mapped[] were going to be passed.
1617 # otherwise an error should have been raised
1618 # by util._extract_mapped_subtype before we got here.
1619 assert expect_annotations_wo_mapped
1621 if isinstance(value, _DCAttributeOptions):
1622 if (
1623 value._has_dataclass_arguments
1624 and not look_for_dataclass_things
1625 ):
1626 if isinstance(value, MapperProperty):
1627 argnames = [
1628 "init",
1629 "default_factory",
1630 "repr",
1631 "default",
1632 "dataclass_metadata",
1633 ]
1634 else:
1635 argnames = [
1636 "init",
1637 "default_factory",
1638 "repr",
1639 "dataclass_metadata",
1640 ]
1642 args = {
1643 a
1644 for a in argnames
1645 if getattr(
1646 value._attribute_options, f"dataclasses_{a}"
1647 )
1648 is not _NoArg.NO_ARG
1649 }
1651 raise exc.ArgumentError(
1652 f"Attribute '{k}' on class {cls} includes "
1653 f"dataclasses argument(s): "
1654 f"{', '.join(sorted(repr(a) for a in args))} but "
1655 f"class does not specify "
1656 "SQLAlchemy native dataclass configuration."
1657 )
1659 if not isinstance(value, (MapperProperty, _MapsColumns)):
1660 # filter for _DCAttributeOptions objects that aren't
1661 # MapperProperty / mapped_column(). Currently this
1662 # includes AssociationProxy. pop it from the things
1663 # we're going to map and set it up as a descriptor
1664 # on the class.
1665 collected_attributes.pop(k)
1667 # Assoc Prox (or other descriptor object that may
1668 # use _DCAttributeOptions) is usually here, except if
1669 # 1. we're a
1670 # dataclass, dataclasses would have removed the
1671 # attr here or 2. assoc proxy is coming from a
1672 # superclass, we want it to be direct here so it
1673 # tracks state or 3. assoc prox comes from
1674 # declared_attr, uncommon case
1675 setattr(cls, k, value)
1676 continue
1678 our_stuff[k] = value
1680 def _extract_declared_columns(self) -> None:
1681 our_stuff = self.properties
1683 # extract columns from the class dict
1684 declared_columns = self.declared_columns
1685 column_ordering = self.column_ordering
1686 name_to_prop_key = collections.defaultdict(set)
1688 for key, c in list(our_stuff.items()):
1689 if isinstance(c, _MapsColumns):
1690 mp_to_assign = c.mapper_property_to_assign
1691 if mp_to_assign:
1692 our_stuff[key] = mp_to_assign
1693 else:
1694 # if no mapper property to assign, this currently means
1695 # this is a MappedColumn that will produce a Column for us
1696 del our_stuff[key]
1698 for col, sort_order in c.columns_to_assign:
1699 if not isinstance(c, CompositeProperty):
1700 name_to_prop_key[col.name].add(key)
1701 declared_columns.add(col)
1703 # we would assert this, however we want the below
1704 # warning to take effect instead. See #9630
1705 # assert col not in column_ordering
1707 column_ordering[col] = sort_order
1709 # if this is a MappedColumn and the attribute key we
1710 # have is not what the column has for its key, map the
1711 # Column explicitly under the attribute key name.
1712 # otherwise, Mapper will map it under the column key.
1713 if mp_to_assign is None and key != col.key:
1714 our_stuff[key] = col
1715 elif isinstance(c, Column):
1716 # undefer previously occurred here, and now occurs earlier.
1717 # ensure every column we get here has been named
1718 assert c.name is not None
1719 name_to_prop_key[c.name].add(key)
1720 declared_columns.add(c)
1721 # if the column is the same name as the key,
1722 # remove it from the explicit properties dict.
1723 # the normal rules for assigning column-based properties
1724 # will take over, including precedence of columns
1725 # in multi-column ColumnProperties.
1726 if key == c.key:
1727 del our_stuff[key]
1729 for name, keys in name_to_prop_key.items():
1730 if len(keys) > 1:
1731 util.warn(
1732 "On class %r, Column object %r named "
1733 "directly multiple times, "
1734 "only one will be used: %s. "
1735 "Consider using orm.synonym instead"
1736 % (self.classname, name, (", ".join(sorted(keys))))
1737 )
1739 def _setup_table(self, table: Optional[FromClause] = None) -> None:
1740 cls = self.cls
1741 cls_as_Decl = cast("MappedClassProtocol[Any]", cls)
1743 tablename = self.tablename
1744 table_args = self.table_args
1745 clsdict_view = self.clsdict_view
1746 declared_columns = self.declared_columns
1747 column_ordering = self.column_ordering
1749 manager = attributes.manager_of_class(cls)
1751 if (
1752 self.table_fn is None
1753 and "__table__" not in clsdict_view
1754 and table is None
1755 ):
1756 if hasattr(cls, "__table_cls__"):
1757 table_cls = cast(
1758 Type[Table],
1759 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore # noqa: E501
1760 )
1761 else:
1762 table_cls = Table
1764 if tablename is not None:
1765 args: Tuple[Any, ...] = ()
1766 table_kw: Dict[str, Any] = {}
1768 if table_args:
1769 if isinstance(table_args, dict):
1770 table_kw = table_args
1771 elif isinstance(table_args, tuple):
1772 if isinstance(table_args[-1], dict):
1773 args, table_kw = table_args[0:-1], table_args[-1]
1774 else:
1775 args = table_args
1777 autoload_with = clsdict_view.get("__autoload_with__")
1778 if autoload_with:
1779 table_kw["autoload_with"] = autoload_with
1781 autoload = clsdict_view.get("__autoload__")
1782 if autoload:
1783 table_kw["autoload"] = True
1785 sorted_columns = sorted(
1786 declared_columns,
1787 key=lambda c: column_ordering.get(c, 0),
1788 )
1789 table = self.set_cls_attribute(
1790 "__table__",
1791 table_cls(
1792 tablename,
1793 self._metadata_for_cls(manager),
1794 *sorted_columns,
1795 *args,
1796 **table_kw,
1797 ),
1798 )
1799 else:
1800 if table is None:
1801 if self.table_fn:
1802 table = self.set_cls_attribute(
1803 "__table__", self.table_fn()
1804 )
1805 else:
1806 table = cls_as_Decl.__table__
1807 if declared_columns:
1808 for c in declared_columns:
1809 if not table.c.contains_column(c):
1810 raise exc.ArgumentError(
1811 "Can't add additional column %r when "
1812 "specifying __table__" % c.key
1813 )
1815 self.local_table = table
1817 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData:
1818 return _metadata_for_cls_fn(self.cls, manager.registry)
1820 def _setup_inheriting_mapper(self) -> None:
1821 cls = self.cls
1823 inherits = None
1825 if inherits is None:
1826 # since we search for classical mappings now, search for
1827 # multiple mapped bases as well and raise an error.
1828 inherits_search = []
1829 for base_ in cls.__bases__:
1830 c = _resolve_for_abstract_or_classical(base_)
1831 if c is None:
1832 continue
1834 if _is_supercls_for_inherits(c) and c not in inherits_search:
1835 inherits_search.append(c)
1837 if inherits_search:
1838 if len(inherits_search) > 1:
1839 raise exc.InvalidRequestError(
1840 "Class %s has multiple mapped bases: %r"
1841 % (cls, inherits_search)
1842 )
1843 inherits = inherits_search[0]
1844 elif isinstance(inherits, Mapper):
1845 inherits = inherits.class_
1847 self.inherits = inherits
1849 clsdict_view = self.clsdict_view
1850 if "__table__" not in clsdict_view and self.tablename is None:
1851 self.single = True
1853 def _setup_inheriting_columns(self) -> None:
1854 table = self.local_table
1855 cls = self.cls
1856 table_args = self.table_args
1857 declared_columns = self.declared_columns
1859 if (
1860 table is None
1861 and self.inherits is None
1862 and not _get_immediate_cls_attr(cls, "__no_table__")
1863 ):
1864 raise exc.InvalidRequestError(
1865 "Class %r does not have a __table__ or __tablename__ "
1866 "specified and does not inherit from an existing "
1867 "table-mapped class." % cls
1868 )
1869 elif self.inherits:
1870 inherited_mapper_or_config = _declared_mapping_info(self.inherits)
1871 assert inherited_mapper_or_config is not None
1872 inherited_table = inherited_mapper_or_config.local_table
1873 inherited_persist_selectable = (
1874 inherited_mapper_or_config.persist_selectable
1875 )
1877 if table is None:
1878 # single table inheritance.
1879 # ensure no table args
1880 if table_args:
1881 raise exc.ArgumentError(
1882 "Can't place __table_args__ on an inherited class "
1883 "with no table."
1884 )
1886 # add any columns declared here to the inherited table.
1887 if declared_columns and not isinstance(inherited_table, Table):
1888 raise exc.ArgumentError(
1889 f"Can't declare columns on single-table-inherited "
1890 f"subclass {self.cls}; superclass {self.inherits} "
1891 "is not mapped to a Table"
1892 )
1894 for col in declared_columns:
1895 assert inherited_table is not None
1896 if col.name in inherited_table.c:
1897 if inherited_table.c[col.name] is col:
1898 continue
1899 raise exc.ArgumentError(
1900 f"Column '{col}' on class {cls.__name__} "
1901 f"conflicts with existing column "
1902 f"'{inherited_table.c[col.name]}'. If using "
1903 f"Declarative, consider using the "
1904 "use_existing_column parameter of mapped_column() "
1905 "to resolve conflicts."
1906 )
1907 if col.primary_key:
1908 raise exc.ArgumentError(
1909 "Can't place primary key columns on an inherited "
1910 "class with no table."
1911 )
1913 if TYPE_CHECKING:
1914 assert isinstance(inherited_table, Table)
1916 inherited_table.append_column(col)
1917 if (
1918 inherited_persist_selectable is not None
1919 and inherited_persist_selectable is not inherited_table
1920 ):
1921 inherited_persist_selectable._refresh_for_new_column(
1922 col
1923 )
1925 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None:
1926 properties = self.properties
1928 if self.mapper_args_fn:
1929 mapper_args = self.mapper_args_fn()
1930 else:
1931 mapper_args = {}
1933 if mapper_kw:
1934 mapper_args.update(mapper_kw)
1936 if "properties" in mapper_args:
1937 properties = dict(properties)
1938 properties.update(mapper_args["properties"])
1940 # make sure that column copies are used rather
1941 # than the original columns from any mixins
1942 for k in ("version_id_col", "polymorphic_on"):
1943 if k in mapper_args:
1944 v = mapper_args[k]
1945 mapper_args[k] = self.column_copies.get(v, v)
1947 if "primary_key" in mapper_args:
1948 mapper_args["primary_key"] = [
1949 self.column_copies.get(v, v)
1950 for v in util.to_list(mapper_args["primary_key"])
1951 ]
1953 if "inherits" in mapper_args:
1954 inherits_arg = mapper_args["inherits"]
1955 if isinstance(inherits_arg, Mapper):
1956 inherits_arg = inherits_arg.class_
1958 if inherits_arg is not self.inherits:
1959 raise exc.InvalidRequestError(
1960 "mapper inherits argument given for non-inheriting "
1961 "class %s" % (mapper_args["inherits"])
1962 )
1964 if self.inherits:
1965 mapper_args["inherits"] = self.inherits
1967 if self.inherits and not mapper_args.get("concrete", False):
1968 # note the superclass is expected to have a Mapper assigned and
1969 # not be a deferred config, as this is called within map()
1970 inherited_mapper = class_mapper(self.inherits, False)
1971 inherited_table = inherited_mapper.local_table
1973 # single or joined inheritance
1974 # exclude any cols on the inherited table which are
1975 # not mapped on the parent class, to avoid
1976 # mapping columns specific to sibling/nephew classes
1977 if "exclude_properties" not in mapper_args:
1978 mapper_args["exclude_properties"] = exclude_properties = {
1979 c.key
1980 for c in inherited_table.c
1981 if c not in inherited_mapper._columntoproperty
1982 }.union(inherited_mapper.exclude_properties or ())
1983 exclude_properties.difference_update(
1984 [c.key for c in self.declared_columns]
1985 )
1987 # look through columns in the current mapper that
1988 # are keyed to a propname different than the colname
1989 # (if names were the same, we'd have popped it out above,
1990 # in which case the mapper makes this combination).
1991 # See if the superclass has a similar column property.
1992 # If so, join them together.
1993 for k, col in list(properties.items()):
1994 if not isinstance(col, expression.ColumnElement):
1995 continue
1996 if k in inherited_mapper._props:
1997 p = inherited_mapper._props[k]
1998 if isinstance(p, ColumnProperty):
1999 # note here we place the subclass column
2000 # first. See [ticket:1892] for background.
2001 properties[k] = [col] + p.columns
2002 result_mapper_args = mapper_args.copy()
2003 result_mapper_args["properties"] = properties
2004 self.mapper_args = result_mapper_args
2006 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2007 self._prepare_mapper_arguments(mapper_kw)
2008 if hasattr(self.cls, "__mapper_cls__"):
2009 mapper_cls = cast(
2010 "Type[Mapper[Any]]",
2011 util.unbound_method_to_callable(
2012 self.cls.__mapper_cls__ # type: ignore
2013 ),
2014 )
2015 else:
2016 mapper_cls = Mapper
2018 return self.set_cls_attribute(
2019 "__mapper__",
2020 mapper_cls(self.cls, self.local_table, **self.mapper_args),
2021 )
2024class _UnmappedDataclassConfig(_ClassScanAbstractConfig):
2025 """Configurator that will produce an unmapped dataclass."""
2027 __slots__ = (
2028 "clsdict_view",
2029 "collected_attributes",
2030 "collected_annotations",
2031 "allow_dataclass_fields",
2032 "dataclass_setup_arguments",
2033 "is_dataclass_prior_to_mapping",
2034 "allow_unmapped_annotations",
2035 )
2037 def __init__(
2038 self,
2039 cls_: Type[_O],
2040 dict_: _ClassDict,
2041 ):
2042 super().__init__(cls_)
2043 self.clsdict_view = (
2044 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
2045 )
2046 self.dataclass_setup_arguments = getattr(
2047 self.cls, "_sa_apply_dc_transforms", None
2048 )
2050 self.is_dataclass_prior_to_mapping = dataclasses.is_dataclass(cls_)
2051 self.allow_dataclass_fields = False
2052 self.allow_unmapped_annotations = True
2053 self.collected_attributes = {}
2054 self.collected_annotations = {}
2056 self._scan_attributes()
2058 self._setup_dataclasses_transforms(
2059 enable_descriptor_defaults=False, revert=True
2060 )
2062 def _scan_attributes(self) -> None:
2063 cls = self.cls
2065 clsdict_view = self.clsdict_view
2066 collected_attributes = self.collected_attributes
2067 _include_dunders = self._include_dunders
2069 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
2071 local_attributes_for_class = self._cls_attr_resolver(cls)
2072 for (
2073 name,
2074 obj,
2075 annotation,
2076 is_dataclass_field,
2077 ) in local_attributes_for_class():
2078 if name in _include_dunders:
2079 continue
2080 elif is_dataclass_field and (
2081 name not in clsdict_view or clsdict_view[name] is not obj
2082 ):
2083 # here, we are definitely looking at the target class
2084 # and not a superclass. this is currently a
2085 # dataclass-only path. if the name is only
2086 # a dataclass field and isn't in local cls.__dict__,
2087 # put the object there.
2088 # assert that the dataclass-enabled resolver agrees
2089 # with what we are seeing
2091 assert not attribute_is_overridden(name, obj)
2093 if _is_declarative_props(obj):
2094 obj = obj.fget()
2096 collected_attributes[name] = obj
2097 self._collect_annotation(name, annotation, cls, False, obj)
2098 else:
2099 self._collect_annotation(name, annotation, cls, None, obj)
2100 if name in clsdict_view:
2101 collected_attributes[name] = obj
2104@util.preload_module("sqlalchemy.orm.decl_api")
2105def _as_dc_declaredattr(
2106 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str
2107) -> Any:
2108 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr.
2109 # we can't write it because field.metadata is immutable :( so we have
2110 # to go through extra trouble to compare these
2111 decl_api = util.preloaded.orm_decl_api
2112 obj = field_metadata[sa_dataclass_metadata_key]
2113 if callable(obj) and not isinstance(obj, decl_api.declared_attr):
2114 return decl_api.declared_attr(obj)
2115 else:
2116 return obj
2119class _DeferredDeclarativeConfig(_DeclarativeMapperConfig):
2120 """Configurator that extends _DeclarativeMapperConfig to add a
2121 "deferred" step, to allow extensions like AbstractConcreteBase,
2122 DeferredMapping to partially set up a mapping that is "prepared"
2123 when table metadata is ready.
2125 """
2127 _cls: weakref.ref[Type[Any]]
2129 is_deferred = True
2131 _configs: util.OrderedDict[
2132 weakref.ref[Type[Any]], _DeferredDeclarativeConfig
2133 ] = util.OrderedDict()
2135 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
2136 pass
2138 @property
2139 def cls(self) -> Type[Any]:
2140 return self._cls() # type: ignore
2142 @cls.setter
2143 def cls(self, class_: Type[Any]) -> None:
2144 self._cls = weakref.ref(class_, self._remove_config_cls)
2145 self._configs[self._cls] = self
2147 @classmethod
2148 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None:
2149 cls._configs.pop(ref, None)
2151 @classmethod
2152 def has_cls(cls, class_: Type[Any]) -> bool:
2153 # 2.6 fails on weakref if class_ is an old style class
2154 return isinstance(class_, type) and weakref.ref(class_) in cls._configs
2156 @classmethod
2157 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn:
2158 if hasattr(class_, "_sa_raise_deferred_config"):
2159 class_._sa_raise_deferred_config()
2161 raise orm_exc.UnmappedClassError(
2162 class_,
2163 msg=(
2164 f"Class {orm_exc._safe_cls_name(class_)} has a deferred "
2165 "mapping on it. It is not yet usable as a mapped class."
2166 ),
2167 )
2169 @classmethod
2170 def config_for_cls(cls, class_: Type[Any]) -> _DeferredDeclarativeConfig:
2171 return cls._configs[weakref.ref(class_)]
2173 @classmethod
2174 def classes_for_base(
2175 cls, base_cls: Type[Any], sort: bool = True
2176 ) -> List[_DeferredDeclarativeConfig]:
2177 classes_for_base = [
2178 m
2179 for m, cls_ in [(m, m.cls) for m in cls._configs.values()]
2180 if cls_ is not None and issubclass(cls_, base_cls)
2181 ]
2183 if not sort:
2184 return classes_for_base
2186 all_m_by_cls = {m.cls: m for m in classes_for_base}
2188 tuples: List[
2189 Tuple[_DeferredDeclarativeConfig, _DeferredDeclarativeConfig]
2190 ] = []
2191 for m_cls in all_m_by_cls:
2192 tuples.extend(
2193 (all_m_by_cls[base_cls], all_m_by_cls[m_cls])
2194 for base_cls in m_cls.__bases__
2195 if base_cls in all_m_by_cls
2196 )
2197 return list(topological.sort(tuples, classes_for_base))
2199 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2200 self._configs.pop(self._cls, None)
2201 return super().map(mapper_kw)
2204def _add_attribute(
2205 cls: Type[Any], key: str, value: MapperProperty[Any]
2206) -> None:
2207 """add an attribute to an existing declarative class.
2209 This runs through the logic to determine MapperProperty,
2210 adds it to the Mapper, adds a column to the mapped Table, etc.
2212 """
2214 if "__mapper__" in cls.__dict__:
2215 mapped_cls = cast("MappedClassProtocol[Any]", cls)
2217 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table:
2218 if isinstance(mc.__table__, Table):
2219 return mc.__table__
2220 raise exc.InvalidRequestError(
2221 f"Cannot add a new attribute to mapped class {mc.__name__!r} "
2222 "because it's not mapped against a table."
2223 )
2225 if isinstance(value, Column):
2226 _undefer_column_name(key, value)
2227 _table_or_raise(mapped_cls).append_column(
2228 value, replace_existing=True
2229 )
2230 mapped_cls.__mapper__.add_property(key, value)
2231 elif isinstance(value, _MapsColumns):
2232 mp = value.mapper_property_to_assign
2233 for col, _ in value.columns_to_assign:
2234 _undefer_column_name(key, col)
2235 _table_or_raise(mapped_cls).append_column(
2236 col, replace_existing=True
2237 )
2238 if not mp:
2239 mapped_cls.__mapper__.add_property(key, col)
2240 if mp:
2241 mapped_cls.__mapper__.add_property(key, mp)
2242 elif isinstance(value, MapperProperty):
2243 mapped_cls.__mapper__.add_property(key, value)
2244 elif isinstance(value, QueryableAttribute) and value.key != key:
2245 # detect a QueryableAttribute that's already mapped being
2246 # assigned elsewhere in userland, turn into a synonym()
2247 value = SynonymProperty(value.key)
2248 mapped_cls.__mapper__.add_property(key, value)
2249 else:
2250 type.__setattr__(cls, key, value)
2251 mapped_cls.__mapper__._expire_memoizations()
2252 else:
2253 type.__setattr__(cls, key, value)
2256def _del_attribute(cls: Type[Any], key: str) -> None:
2257 if (
2258 "__mapper__" in cls.__dict__
2259 and key in cls.__dict__
2260 and not cast(
2261 "MappedClassProtocol[Any]", cls
2262 ).__mapper__._dispose_called
2263 ):
2264 value = cls.__dict__[key]
2265 if isinstance(
2266 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute)
2267 ):
2268 raise NotImplementedError(
2269 "Can't un-map individual mapped attributes on a mapped class."
2270 )
2271 else:
2272 type.__delattr__(cls, key)
2273 cast(
2274 "MappedClassProtocol[Any]", cls
2275 ).__mapper__._expire_memoizations()
2276 else:
2277 type.__delattr__(cls, key)
2280def _declarative_constructor(self: Any, **kwargs: Any) -> None:
2281 """A simple constructor that allows initialization from kwargs.
2283 Sets attributes on the constructed instance using the names and
2284 values in ``kwargs``.
2286 Only keys that are present as
2287 attributes of the instance's class are allowed. These could be,
2288 for example, any mapped columns or relationships.
2289 """
2290 cls_ = type(self)
2291 for k in kwargs:
2292 if not hasattr(cls_, k):
2293 raise TypeError(
2294 "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
2295 )
2296 setattr(self, k, kwargs[k])
2299_declarative_constructor.__name__ = "__init__"
2302def _undefer_column_name(key: str, column: Column[Any]) -> None:
2303 if column.key is None:
2304 column.key = key
2305 if column.name is None:
2306 column.name = key