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