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 class_mapper
62from .util import de_stringify_annotation
63from .. import event
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 finally:
665 if revert and revert_dict:
666 # used for mixin dataclasses; we have to restore the
667 # mapped_column(), relationship() etc. to the class so these
668 # take place for a mapped class scan
669 for k, v in revert_dict.items():
670 setattr(self.cls, k, v)
672 restore_anno()
674 def _collect_annotation(
675 self,
676 name: str,
677 raw_annotation: _AnnotationScanType,
678 originating_class: Type[Any],
679 expect_mapped: Optional[bool],
680 attr_value: Any,
681 ) -> Optional[_CollectedAnnotation]:
682 if name in self.collected_annotations:
683 return self.collected_annotations[name]
685 if raw_annotation is None:
686 return None
688 is_dataclass = self.is_dataclass_prior_to_mapping
689 allow_unmapped = self.allow_unmapped_annotations
691 if expect_mapped is None:
692 is_dataclass_field = isinstance(attr_value, dataclasses.Field)
693 expect_mapped = (
694 not is_dataclass_field
695 and not allow_unmapped
696 and (
697 attr_value is None
698 or isinstance(attr_value, _MappedAttribute)
699 )
700 )
702 is_dataclass_field = False
703 extracted = _extract_mapped_subtype(
704 raw_annotation,
705 self.cls,
706 originating_class.__module__,
707 name,
708 type(attr_value),
709 required=False,
710 is_dataclass_field=is_dataclass_field,
711 expect_mapped=expect_mapped and not is_dataclass,
712 )
713 if extracted is None:
714 # ClassVar can come out here
715 return None
717 extracted_mapped_annotation, mapped_container = extracted
719 if attr_value is None and not is_literal(extracted_mapped_annotation):
720 for elem in get_args(extracted_mapped_annotation):
721 if is_fwd_ref(
722 elem, check_generic=True, check_for_plain_string=True
723 ):
724 elem = de_stringify_annotation(
725 self.cls,
726 elem,
727 originating_class.__module__,
728 include_generic=True,
729 )
730 # look in Annotated[...] for an ORM construct,
731 # such as Annotated[int, mapped_column(primary_key=True)]
732 if isinstance(elem, _IntrospectsAnnotations):
733 attr_value = elem.found_in_pep593_annotated()
735 self.collected_annotations[name] = ca = _CollectedAnnotation(
736 raw_annotation,
737 mapped_container,
738 extracted_mapped_annotation,
739 is_dataclass,
740 attr_value,
741 originating_class.__module__,
742 originating_class,
743 )
744 return ca
746 @classmethod
747 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None:
748 allowed = {
749 "init",
750 "repr",
751 "order",
752 "eq",
753 "unsafe_hash",
754 "kw_only",
755 "match_args",
756 "dataclass_callable",
757 }
758 disallowed_args = set(arguments).difference(allowed)
759 if disallowed_args:
760 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args))
761 raise exc.ArgumentError(
762 f"Dataclass argument(s) {msg} are not accepted"
763 )
765 def _cls_attr_override_checker(
766 self, cls: Type[_O]
767 ) -> Callable[[str, Any], bool]:
768 """Produce a function that checks if a class has overridden an
769 attribute, taking SQLAlchemy-enabled dataclass fields into account.
771 """
773 if self.allow_dataclass_fields:
774 sa_dataclass_metadata_key = _get_immediate_cls_attr(
775 cls, "__sa_dataclass_metadata_key__"
776 )
777 else:
778 sa_dataclass_metadata_key = None
780 if not sa_dataclass_metadata_key:
782 def attribute_is_overridden(key: str, obj: Any) -> bool:
783 return getattr(cls, key, obj) is not obj
785 else:
786 all_datacls_fields = {
787 f.name: f.metadata[sa_dataclass_metadata_key]
788 for f in util.dataclass_fields(cls)
789 if sa_dataclass_metadata_key in f.metadata
790 }
791 local_datacls_fields = {
792 f.name: f.metadata[sa_dataclass_metadata_key]
793 for f in util.local_dataclass_fields(cls)
794 if sa_dataclass_metadata_key in f.metadata
795 }
797 absent = object()
799 def attribute_is_overridden(key: str, obj: Any) -> bool:
800 if _is_declarative_props(obj):
801 obj = obj.fget
803 # this function likely has some failure modes still if
804 # someone is doing a deep mixing of the same attribute
805 # name as plain Python attribute vs. dataclass field.
807 ret = local_datacls_fields.get(key, absent)
808 if _is_declarative_props(ret):
809 ret = ret.fget
811 if ret is obj:
812 return False
813 elif ret is not absent:
814 return True
816 all_field = all_datacls_fields.get(key, absent)
818 ret = getattr(cls, key, obj)
820 if ret is obj:
821 return False
823 # for dataclasses, this could be the
824 # 'default' of the field. so filter more specifically
825 # for an already-mapped InstrumentedAttribute
826 if ret is not absent and isinstance(
827 ret, InstrumentedAttribute
828 ):
829 return True
831 if all_field is obj:
832 return False
833 elif all_field is not absent:
834 return True
836 # can't find another attribute
837 return False
839 return attribute_is_overridden
841 def _cls_attr_resolver(
842 self, cls: Type[Any]
843 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
844 """produce a function to iterate the "attributes" of a class
845 which we want to consider for mapping, adjusting for SQLAlchemy fields
846 embedded in dataclass fields.
848 """
849 cls_annotations = util.get_annotations(cls)
851 cls_vars = vars(cls)
853 _include_dunders = self._include_dunders
854 _match_exclude_dunders = self._match_exclude_dunders
856 names = [
857 n
858 for n in util.merge_lists_w_ordering(
859 list(cls_vars), list(cls_annotations)
860 )
861 if not _match_exclude_dunders.match(n) or n in _include_dunders
862 ]
864 if self.allow_dataclass_fields:
865 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
866 cls, "__sa_dataclass_metadata_key__"
867 )
868 else:
869 sa_dataclass_metadata_key = None
871 if not sa_dataclass_metadata_key:
873 def local_attributes_for_class() -> (
874 Iterable[Tuple[str, Any, Any, bool]]
875 ):
876 return (
877 (
878 name,
879 cls_vars.get(name),
880 cls_annotations.get(name),
881 False,
882 )
883 for name in names
884 )
886 else:
887 dataclass_fields = {
888 field.name: field for field in util.local_dataclass_fields(cls)
889 }
891 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
893 def local_attributes_for_class() -> (
894 Iterable[Tuple[str, Any, Any, bool]]
895 ):
896 for name in names:
897 field = dataclass_fields.get(name, None)
898 if field and sa_dataclass_metadata_key in field.metadata:
899 yield field.name, _as_dc_declaredattr(
900 field.metadata, fixed_sa_dataclass_metadata_key
901 ), cls_annotations.get(field.name), True
902 else:
903 yield name, cls_vars.get(name), cls_annotations.get(
904 name
905 ), False
907 return local_attributes_for_class
910class _DeclarativeMapperConfig(_MapperConfig, _ClassScanAbstractConfig):
911 """Configurator that will produce a declarative mapped class"""
913 __slots__ = (
914 "registry",
915 "local_table",
916 "persist_selectable",
917 "declared_columns",
918 "column_ordering",
919 "column_copies",
920 "table_args",
921 "tablename",
922 "mapper_args",
923 "mapper_args_fn",
924 "table_fn",
925 "inherits",
926 "single",
927 "clsdict_view",
928 "collected_attributes",
929 "collected_annotations",
930 "allow_dataclass_fields",
931 "dataclass_setup_arguments",
932 "is_dataclass_prior_to_mapping",
933 "allow_unmapped_annotations",
934 )
936 is_deferred = False
937 registry: _RegistryType
938 local_table: Optional[FromClause]
939 persist_selectable: Optional[FromClause]
940 declared_columns: util.OrderedSet[Column[Any]]
941 column_ordering: Dict[Column[Any], int]
942 column_copies: Dict[
943 Union[MappedColumn[Any], Column[Any]],
944 Union[MappedColumn[Any], Column[Any]],
945 ]
946 tablename: Optional[str]
947 mapper_args: Mapping[str, Any]
948 table_args: Optional[_TableArgsType]
949 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
950 inherits: Optional[Type[Any]]
951 single: bool
953 def __init__(
954 self,
955 registry: _RegistryType,
956 cls_: Type[_O],
957 dict_: _ClassDict,
958 ):
959 # grab class dict before the instrumentation manager has been added.
960 # reduces cycles
961 self.clsdict_view = (
962 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
963 )
964 super().__init__(registry, cls_)
965 self.registry = registry
966 self.persist_selectable = None
968 self.collected_attributes = {}
969 self.collected_annotations = {}
970 self.declared_columns = util.OrderedSet()
971 self.column_ordering = {}
972 self.column_copies = {}
973 self.single = False
974 self.dataclass_setup_arguments = dca = getattr(
975 self.cls, "_sa_apply_dc_transforms", None
976 )
978 self.allow_unmapped_annotations = getattr(
979 self.cls, "__allow_unmapped__", False
980 ) or bool(self.dataclass_setup_arguments)
982 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
983 cls_
984 )
986 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
988 # we don't want to consume Field objects from a not-already-dataclass.
989 # the Field objects won't have their "name" or "type" populated,
990 # and while it seems like we could just set these on Field as we
991 # read them, Field is documented as "user read only" and we need to
992 # stay far away from any off-label use of dataclasses APIs.
993 if (not cld or dca) and sdk:
994 raise exc.InvalidRequestError(
995 "SQLAlchemy mapped dataclasses can't consume mapping "
996 "information from dataclass.Field() objects if the immediate "
997 "class is not already a dataclass."
998 )
1000 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
1001 # then also look inside of dataclass.Field() objects yielded by
1002 # dataclasses.get_fields(cls) when scanning for attributes
1003 self.allow_dataclass_fields = bool(sdk and cld)
1005 self._setup_declared_events()
1007 self._scan_attributes()
1009 self._setup_dataclasses_transforms(enable_descriptor_defaults=True)
1011 with mapperlib._CONFIGURE_MUTEX:
1012 clsregistry._add_class(
1013 self.classname, self.cls, registry._class_registry
1014 )
1016 self._setup_inheriting_mapper()
1018 self._extract_mappable_attributes()
1020 self._extract_declared_columns()
1022 self._setup_table()
1024 self._setup_inheriting_columns()
1026 self._early_mapping(util.EMPTY_DICT)
1028 def _setup_declared_events(self) -> None:
1029 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
1031 @event.listens_for(Mapper, "after_configured")
1032 def after_configured() -> None:
1033 cast(
1034 "_DeclMappedClassProtocol[Any]", self.cls
1035 ).__declare_last__()
1037 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
1039 @event.listens_for(Mapper, "before_configured")
1040 def before_configured() -> None:
1041 cast(
1042 "_DeclMappedClassProtocol[Any]", self.cls
1043 ).__declare_first__()
1045 def _scan_attributes(self) -> None:
1046 cls = self.cls
1048 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
1050 clsdict_view = self.clsdict_view
1051 collected_attributes = self.collected_attributes
1052 column_copies = self.column_copies
1053 _include_dunders = self._include_dunders
1054 mapper_args_fn = None
1055 table_args = inherited_table_args = None
1056 table_fn = None
1057 tablename = None
1058 fixed_table = "__table__" in clsdict_view
1060 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
1062 bases = []
1064 for base in cls.__mro__:
1065 # collect bases and make sure standalone columns are copied
1066 # to be the column they will ultimately be on the class,
1067 # so that declared_attr functions use the right columns.
1068 # need to do this all the way up the hierarchy first
1069 # (see #8190)
1071 class_mapped = base is not cls and _is_supercls_for_inherits(base)
1073 local_attributes_for_class = self._cls_attr_resolver(base)
1075 if not class_mapped and base is not cls:
1076 locally_collected_columns = self._produce_column_copies(
1077 local_attributes_for_class,
1078 attribute_is_overridden,
1079 fixed_table,
1080 base,
1081 )
1082 else:
1083 locally_collected_columns = {}
1085 bases.append(
1086 (
1087 base,
1088 class_mapped,
1089 local_attributes_for_class,
1090 locally_collected_columns,
1091 )
1092 )
1094 for (
1095 base,
1096 class_mapped,
1097 local_attributes_for_class,
1098 locally_collected_columns,
1099 ) in bases:
1100 # this transfer can also take place as we scan each name
1101 # for finer-grained control of how collected_attributes is
1102 # populated, as this is what impacts column ordering.
1103 # however it's simpler to get it out of the way here.
1104 collected_attributes.update(locally_collected_columns)
1106 for (
1107 name,
1108 obj,
1109 annotation,
1110 is_dataclass_field,
1111 ) in local_attributes_for_class():
1112 if name in _include_dunders:
1113 if name == "__mapper_args__":
1114 check_decl = _check_declared_props_nocascade(
1115 obj, name, cls
1116 )
1117 if not mapper_args_fn and (
1118 not class_mapped or check_decl
1119 ):
1120 # don't even invoke __mapper_args__ until
1121 # after we've determined everything about the
1122 # mapped table.
1123 # make a copy of it so a class-level dictionary
1124 # is not overwritten when we update column-based
1125 # arguments.
1126 def _mapper_args_fn() -> Dict[str, Any]:
1127 return dict(cls_as_Decl.__mapper_args__)
1129 mapper_args_fn = _mapper_args_fn
1131 elif name == "__tablename__":
1132 check_decl = _check_declared_props_nocascade(
1133 obj, name, cls
1134 )
1135 if not tablename and (not class_mapped or check_decl):
1136 tablename = cls_as_Decl.__tablename__
1137 elif name == "__table__":
1138 check_decl = _check_declared_props_nocascade(
1139 obj, name, cls
1140 )
1141 # if a @declared_attr using "__table__" is detected,
1142 # wrap up a callable to look for "__table__" from
1143 # the final concrete class when we set up a table.
1144 # this was fixed by
1145 # #11509, regression in 2.0 from version 1.4.
1146 if check_decl and not table_fn:
1147 # don't even invoke __table__ until we're ready
1148 def _table_fn() -> FromClause:
1149 return cls_as_Decl.__table__
1151 table_fn = _table_fn
1153 elif name == "__table_args__":
1154 check_decl = _check_declared_props_nocascade(
1155 obj, name, cls
1156 )
1157 if not table_args and (not class_mapped or check_decl):
1158 table_args = cls_as_Decl.__table_args__
1159 if not isinstance(
1160 table_args, (tuple, dict, type(None))
1161 ):
1162 raise exc.ArgumentError(
1163 "__table_args__ value must be a tuple, "
1164 "dict, or None"
1165 )
1166 if base is not cls:
1167 inherited_table_args = True
1168 else:
1169 # any other dunder names; should not be here
1170 # as we have tested for all four names in
1171 # _include_dunders
1172 assert False
1173 elif class_mapped:
1174 if _is_declarative_props(obj) and not obj._quiet:
1175 util.warn(
1176 "Regular (i.e. not __special__) "
1177 "attribute '%s.%s' uses @declared_attr, "
1178 "but owning class %s is mapped - "
1179 "not applying to subclass %s."
1180 % (base.__name__, name, base, cls)
1181 )
1183 continue
1184 elif base is not cls:
1185 # we're a mixin, abstract base, or something that is
1186 # acting like that for now.
1188 if isinstance(obj, (Column, MappedColumn)):
1189 # already copied columns to the mapped class.
1190 continue
1191 elif isinstance(obj, MapperProperty):
1192 raise exc.InvalidRequestError(
1193 "Mapper properties (i.e. deferred,"
1194 "column_property(), relationship(), etc.) must "
1195 "be declared as @declared_attr callables "
1196 "on declarative mixin classes. For dataclass "
1197 "field() objects, use a lambda:"
1198 )
1199 elif _is_declarative_props(obj):
1200 # tried to get overloads to tell this to
1201 # pylance, no luck
1202 assert obj is not None
1204 if obj._cascading:
1205 if name in clsdict_view:
1206 # unfortunately, while we can use the user-
1207 # defined attribute here to allow a clean
1208 # override, if there's another
1209 # subclass below then it still tries to use
1210 # this. not sure if there is enough
1211 # information here to add this as a feature
1212 # later on.
1213 util.warn(
1214 "Attribute '%s' on class %s cannot be "
1215 "processed due to "
1216 "@declared_attr.cascading; "
1217 "skipping" % (name, cls)
1218 )
1219 collected_attributes[name] = column_copies[obj] = (
1220 ret
1221 ) = obj.__get__(obj, cls)
1222 setattr(cls, name, ret)
1223 else:
1224 if is_dataclass_field:
1225 # access attribute using normal class access
1226 # first, to see if it's been mapped on a
1227 # superclass. note if the dataclasses.field()
1228 # has "default", this value can be anything.
1229 ret = getattr(cls, name, None)
1231 # so, if it's anything that's not ORM
1232 # mapped, assume we should invoke the
1233 # declared_attr
1234 if not isinstance(ret, InspectionAttr):
1235 ret = obj.fget()
1236 else:
1237 # access attribute using normal class access.
1238 # if the declared attr already took place
1239 # on a superclass that is mapped, then
1240 # this is no longer a declared_attr, it will
1241 # be the InstrumentedAttribute
1242 ret = getattr(cls, name)
1244 # correct for proxies created from hybrid_property
1245 # or similar. note there is no known case that
1246 # produces nested proxies, so we are only
1247 # looking one level deep right now.
1249 if (
1250 isinstance(ret, InspectionAttr)
1251 and attr_is_internal_proxy(ret)
1252 and not isinstance(
1253 ret.original_property, MapperProperty
1254 )
1255 ):
1256 ret = ret.descriptor
1258 collected_attributes[name] = column_copies[obj] = (
1259 ret
1260 )
1262 if (
1263 isinstance(ret, (Column, MapperProperty))
1264 and ret.doc is None
1265 ):
1266 ret.doc = obj.__doc__
1268 self._collect_annotation(
1269 name,
1270 obj._collect_return_annotation(),
1271 base,
1272 True,
1273 obj,
1274 )
1275 elif _is_mapped_annotation(annotation, cls, base):
1276 # Mapped annotation without any object.
1277 # product_column_copies should have handled this.
1278 # if future support for other MapperProperty,
1279 # then test if this name is already handled and
1280 # otherwise proceed to generate.
1281 if not fixed_table:
1282 assert (
1283 name in collected_attributes
1284 or attribute_is_overridden(name, None)
1285 )
1286 continue
1287 else:
1288 # here, the attribute is some other kind of
1289 # property that we assume is not part of the
1290 # declarative mapping. however, check for some
1291 # more common mistakes
1292 self._warn_for_decl_attributes(base, name, obj)
1293 elif is_dataclass_field and (
1294 name not in clsdict_view or clsdict_view[name] is not obj
1295 ):
1296 # here, we are definitely looking at the target class
1297 # and not a superclass. this is currently a
1298 # dataclass-only path. if the name is only
1299 # a dataclass field and isn't in local cls.__dict__,
1300 # put the object there.
1301 # assert that the dataclass-enabled resolver agrees
1302 # with what we are seeing
1304 assert not attribute_is_overridden(name, obj)
1306 if _is_declarative_props(obj):
1307 obj = obj.fget()
1309 collected_attributes[name] = obj
1310 self._collect_annotation(
1311 name, annotation, base, False, obj
1312 )
1313 else:
1314 collected_annotation = self._collect_annotation(
1315 name, annotation, base, None, obj
1316 )
1317 is_mapped = (
1318 collected_annotation is not None
1319 and collected_annotation.mapped_container is not None
1320 )
1321 generated_obj = (
1322 collected_annotation.attr_value
1323 if collected_annotation is not None
1324 else obj
1325 )
1326 if obj is None and not fixed_table and is_mapped:
1327 collected_attributes[name] = (
1328 generated_obj
1329 if generated_obj is not None
1330 else MappedColumn()
1331 )
1332 elif name in clsdict_view:
1333 collected_attributes[name] = obj
1334 # else if the name is not in the cls.__dict__,
1335 # don't collect it as an attribute.
1336 # we will see the annotation only, which is meaningful
1337 # both for mapping and dataclasses setup
1339 if inherited_table_args and not tablename:
1340 table_args = None
1342 self.table_args = table_args
1343 self.tablename = tablename
1344 self.mapper_args_fn = mapper_args_fn
1345 self.table_fn = table_fn
1347 @classmethod
1348 def _update_annotations_for_non_mapped_class(
1349 cls, klass: Type[_O]
1350 ) -> Mapping[str, _AnnotationScanType]:
1351 cls_annotations = util.get_annotations(klass)
1353 new_anno = {}
1354 for name, annotation in cls_annotations.items():
1355 if _is_mapped_annotation(annotation, klass, klass):
1356 extracted = _extract_mapped_subtype(
1357 annotation,
1358 klass,
1359 klass.__module__,
1360 name,
1361 type(None),
1362 required=False,
1363 is_dataclass_field=False,
1364 expect_mapped=False,
1365 )
1366 if extracted:
1367 inner, _ = extracted
1368 new_anno[name] = inner
1369 else:
1370 new_anno[name] = annotation
1371 return new_anno
1373 def _warn_for_decl_attributes(
1374 self, cls: Type[Any], key: str, c: Any
1375 ) -> None:
1376 if isinstance(c, expression.ColumnElement):
1377 util.warn(
1378 f"Attribute '{key}' on class {cls} appears to "
1379 "be a non-schema SQLAlchemy expression "
1380 "object; this won't be part of the declarative mapping. "
1381 "To map arbitrary expressions, use ``column_property()`` "
1382 "or a similar function such as ``deferred()``, "
1383 "``query_expression()`` etc. "
1384 )
1386 def _produce_column_copies(
1387 self,
1388 attributes_for_class: Callable[
1389 [], Iterable[Tuple[str, Any, Any, bool]]
1390 ],
1391 attribute_is_overridden: Callable[[str, Any], bool],
1392 fixed_table: bool,
1393 originating_class: Type[Any],
1394 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]:
1395 cls = self.cls
1396 dict_ = self.clsdict_view
1397 locally_collected_attributes = {}
1398 column_copies = self.column_copies
1399 # copy mixin columns to the mapped class
1401 for name, obj, annotation, is_dataclass in attributes_for_class():
1402 if (
1403 not fixed_table
1404 and obj is None
1405 and _is_mapped_annotation(annotation, cls, originating_class)
1406 ):
1407 # obj is None means this is the annotation only path
1409 if attribute_is_overridden(name, obj):
1410 # perform same "overridden" check as we do for
1411 # Column/MappedColumn, this is how a mixin col is not
1412 # applied to an inherited subclass that does not have
1413 # the mixin. the anno-only path added here for
1414 # #9564
1415 continue
1417 collected_annotation = self._collect_annotation(
1418 name, annotation, originating_class, True, obj
1419 )
1420 obj = (
1421 collected_annotation.attr_value
1422 if collected_annotation is not None
1423 else obj
1424 )
1425 if obj is None:
1426 obj = MappedColumn()
1428 locally_collected_attributes[name] = obj
1429 setattr(cls, name, obj)
1431 elif isinstance(obj, (Column, MappedColumn)):
1432 if attribute_is_overridden(name, obj):
1433 # if column has been overridden
1434 # (like by the InstrumentedAttribute of the
1435 # superclass), skip. don't collect the annotation
1436 # either (issue #8718)
1437 continue
1439 collected_annotation = self._collect_annotation(
1440 name, annotation, originating_class, True, obj
1441 )
1442 obj = (
1443 collected_annotation.attr_value
1444 if collected_annotation is not None
1445 else obj
1446 )
1448 if name not in dict_ and not (
1449 "__table__" in dict_
1450 and (getattr(obj, "name", None) or name)
1451 in dict_["__table__"].c
1452 ):
1453 if obj.foreign_keys:
1454 for fk in obj.foreign_keys:
1455 if (
1456 fk._table_column is not None
1457 and fk._table_column.table is None
1458 ):
1459 raise exc.InvalidRequestError(
1460 "Columns with foreign keys to "
1461 "non-table-bound "
1462 "columns must be declared as "
1463 "@declared_attr callables "
1464 "on declarative mixin classes. "
1465 "For dataclass "
1466 "field() objects, use a lambda:."
1467 )
1469 column_copies[obj] = copy_ = obj._copy()
1471 locally_collected_attributes[name] = copy_
1472 setattr(cls, name, copy_)
1474 return locally_collected_attributes
1476 def _extract_mappable_attributes(self) -> None:
1477 cls = self.cls
1478 collected_attributes = self.collected_attributes
1480 our_stuff = self.properties
1482 _include_dunders = self._include_dunders
1484 late_mapped = _get_immediate_cls_attr(
1485 cls, "_sa_decl_prepare_nocascade", strict=True
1486 )
1488 allow_unmapped_annotations = self.allow_unmapped_annotations
1489 expect_annotations_wo_mapped = (
1490 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping
1491 )
1493 look_for_dataclass_things = bool(self.dataclass_setup_arguments)
1495 for k in list(collected_attributes):
1496 if k in _include_dunders:
1497 continue
1499 value = collected_attributes[k]
1501 if _is_declarative_props(value):
1502 # @declared_attr in collected_attributes only occurs here for a
1503 # @declared_attr that's directly on the mapped class;
1504 # for a mixin, these have already been evaluated
1505 if value._cascading:
1506 util.warn(
1507 "Use of @declared_attr.cascading only applies to "
1508 "Declarative 'mixin' and 'abstract' classes. "
1509 "Currently, this flag is ignored on mapped class "
1510 "%s" % self.cls
1511 )
1513 value = getattr(cls, k)
1515 elif (
1516 isinstance(value, QueryableAttribute)
1517 and value.class_ is not cls
1518 and value.key != k
1519 ):
1520 # detect a QueryableAttribute that's already mapped being
1521 # assigned elsewhere in userland, turn into a synonym()
1522 value = SynonymProperty(value.key)
1523 setattr(cls, k, value)
1525 if (
1526 isinstance(value, tuple)
1527 and len(value) == 1
1528 and isinstance(value[0], (Column, _MappedAttribute))
1529 ):
1530 util.warn(
1531 "Ignoring declarative-like tuple value of attribute "
1532 "'%s': possibly a copy-and-paste error with a comma "
1533 "accidentally placed at the end of the line?" % k
1534 )
1535 continue
1536 elif look_for_dataclass_things and isinstance(
1537 value, dataclasses.Field
1538 ):
1539 # we collected a dataclass Field; dataclasses would have
1540 # set up the correct state on the class
1541 continue
1542 elif not isinstance(value, (Column, _DCAttributeOptions)):
1543 # using @declared_attr for some object that
1544 # isn't Column/MapperProperty/_DCAttributeOptions; remove
1545 # from the clsdict_view
1546 # and place the evaluated value onto the class.
1547 collected_attributes.pop(k)
1548 self._warn_for_decl_attributes(cls, k, value)
1549 if not late_mapped:
1550 setattr(cls, k, value)
1551 continue
1552 # we expect to see the name 'metadata' in some valid cases;
1553 # however at this point we see it's assigned to something trying
1554 # to be mapped, so raise for that.
1555 # TODO: should "registry" here be also? might be too late
1556 # to change that now (2.0 betas)
1557 elif k in ("metadata",):
1558 raise exc.InvalidRequestError(
1559 f"Attribute name '{k}' is reserved when using the "
1560 "Declarative API."
1561 )
1562 elif isinstance(value, Column):
1563 _undefer_column_name(
1564 k, self.column_copies.get(value, value) # type: ignore
1565 )
1566 else:
1567 if isinstance(value, _IntrospectsAnnotations):
1568 (
1569 annotation,
1570 mapped_container,
1571 extracted_mapped_annotation,
1572 is_dataclass,
1573 attr_value,
1574 originating_module,
1575 originating_class,
1576 ) = self.collected_annotations.get(
1577 k, (None, None, None, False, None, None, None)
1578 )
1580 # issue #8692 - don't do any annotation interpretation if
1581 # an annotation were present and a container such as
1582 # Mapped[] etc. were not used. If annotation is None,
1583 # do declarative_scan so that the property can raise
1584 # for required
1585 if (
1586 mapped_container is not None
1587 or annotation is None
1588 # issue #10516: need to do declarative_scan even with
1589 # a non-Mapped annotation if we are doing
1590 # __allow_unmapped__, for things like col.name
1591 # assignment
1592 or allow_unmapped_annotations
1593 ):
1594 try:
1595 value.declarative_scan(
1596 self,
1597 self.registry,
1598 cls,
1599 originating_module,
1600 k,
1601 mapped_container,
1602 annotation,
1603 extracted_mapped_annotation,
1604 is_dataclass,
1605 )
1606 except NameError as ne:
1607 raise orm_exc.MappedAnnotationError(
1608 f"Could not resolve all types within mapped "
1609 f'annotation: "{annotation}". Ensure all '
1610 f"types are written correctly and are "
1611 f"imported within the module in use."
1612 ) from ne
1613 else:
1614 # assert that we were expecting annotations
1615 # without Mapped[] were going to be passed.
1616 # otherwise an error should have been raised
1617 # by util._extract_mapped_subtype before we got here.
1618 assert expect_annotations_wo_mapped
1620 if isinstance(value, _DCAttributeOptions):
1621 if (
1622 value._has_dataclass_arguments
1623 and not look_for_dataclass_things
1624 ):
1625 if isinstance(value, MapperProperty):
1626 argnames = [
1627 "init",
1628 "default_factory",
1629 "repr",
1630 "default",
1631 "dataclass_metadata",
1632 ]
1633 else:
1634 argnames = [
1635 "init",
1636 "default_factory",
1637 "repr",
1638 "dataclass_metadata",
1639 ]
1641 args = {
1642 a
1643 for a in argnames
1644 if getattr(
1645 value._attribute_options, f"dataclasses_{a}"
1646 )
1647 is not _NoArg.NO_ARG
1648 }
1650 raise exc.ArgumentError(
1651 f"Attribute '{k}' on class {cls} includes "
1652 f"dataclasses argument(s): "
1653 f"{', '.join(sorted(repr(a) for a in args))} but "
1654 f"class does not specify "
1655 "SQLAlchemy native dataclass configuration."
1656 )
1658 if not isinstance(value, (MapperProperty, _MapsColumns)):
1659 # filter for _DCAttributeOptions objects that aren't
1660 # MapperProperty / mapped_column(). Currently this
1661 # includes AssociationProxy. pop it from the things
1662 # we're going to map and set it up as a descriptor
1663 # on the class.
1664 collected_attributes.pop(k)
1666 # Assoc Prox (or other descriptor object that may
1667 # use _DCAttributeOptions) is usually here, except if
1668 # 1. we're a
1669 # dataclass, dataclasses would have removed the
1670 # attr here or 2. assoc proxy is coming from a
1671 # superclass, we want it to be direct here so it
1672 # tracks state or 3. assoc prox comes from
1673 # declared_attr, uncommon case
1674 setattr(cls, k, value)
1675 continue
1677 our_stuff[k] = value
1679 def _extract_declared_columns(self) -> None:
1680 our_stuff = self.properties
1682 # extract columns from the class dict
1683 declared_columns = self.declared_columns
1684 column_ordering = self.column_ordering
1685 name_to_prop_key = collections.defaultdict(set)
1687 for key, c in list(our_stuff.items()):
1688 if isinstance(c, _MapsColumns):
1689 mp_to_assign = c.mapper_property_to_assign
1690 if mp_to_assign:
1691 our_stuff[key] = mp_to_assign
1692 else:
1693 # if no mapper property to assign, this currently means
1694 # this is a MappedColumn that will produce a Column for us
1695 del our_stuff[key]
1697 for col, sort_order in c.columns_to_assign:
1698 if not isinstance(c, CompositeProperty):
1699 name_to_prop_key[col.name].add(key)
1700 declared_columns.add(col)
1702 # we would assert this, however we want the below
1703 # warning to take effect instead. See #9630
1704 # assert col not in column_ordering
1706 column_ordering[col] = sort_order
1708 # if this is a MappedColumn and the attribute key we
1709 # have is not what the column has for its key, map the
1710 # Column explicitly under the attribute key name.
1711 # otherwise, Mapper will map it under the column key.
1712 if mp_to_assign is None and key != col.key:
1713 our_stuff[key] = col
1714 elif isinstance(c, Column):
1715 # undefer previously occurred here, and now occurs earlier.
1716 # ensure every column we get here has been named
1717 assert c.name is not None
1718 name_to_prop_key[c.name].add(key)
1719 declared_columns.add(c)
1720 # if the column is the same name as the key,
1721 # remove it from the explicit properties dict.
1722 # the normal rules for assigning column-based properties
1723 # will take over, including precedence of columns
1724 # in multi-column ColumnProperties.
1725 if key == c.key:
1726 del our_stuff[key]
1728 for name, keys in name_to_prop_key.items():
1729 if len(keys) > 1:
1730 util.warn(
1731 "On class %r, Column object %r named "
1732 "directly multiple times, "
1733 "only one will be used: %s. "
1734 "Consider using orm.synonym instead"
1735 % (self.classname, name, (", ".join(sorted(keys))))
1736 )
1738 def _setup_table(self, table: Optional[FromClause] = None) -> None:
1739 cls = self.cls
1740 cls_as_Decl = cast("MappedClassProtocol[Any]", cls)
1742 tablename = self.tablename
1743 table_args = self.table_args
1744 clsdict_view = self.clsdict_view
1745 declared_columns = self.declared_columns
1746 column_ordering = self.column_ordering
1748 manager = attributes.manager_of_class(cls)
1750 if (
1751 self.table_fn is None
1752 and "__table__" not in clsdict_view
1753 and table is None
1754 ):
1755 if hasattr(cls, "__table_cls__"):
1756 table_cls = cast(
1757 Type[Table],
1758 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore # noqa: E501
1759 )
1760 else:
1761 table_cls = Table
1763 if tablename is not None:
1764 args: Tuple[Any, ...] = ()
1765 table_kw: Dict[str, Any] = {}
1767 if table_args:
1768 if isinstance(table_args, dict):
1769 table_kw = table_args
1770 elif isinstance(table_args, tuple):
1771 if isinstance(table_args[-1], dict):
1772 args, table_kw = table_args[0:-1], table_args[-1]
1773 else:
1774 args = table_args
1776 autoload_with = clsdict_view.get("__autoload_with__")
1777 if autoload_with:
1778 table_kw["autoload_with"] = autoload_with
1780 autoload = clsdict_view.get("__autoload__")
1781 if autoload:
1782 table_kw["autoload"] = True
1784 sorted_columns = sorted(
1785 declared_columns,
1786 key=lambda c: column_ordering.get(c, 0),
1787 )
1788 table = self.set_cls_attribute(
1789 "__table__",
1790 table_cls(
1791 tablename,
1792 self._metadata_for_cls(manager),
1793 *sorted_columns,
1794 *args,
1795 **table_kw,
1796 ),
1797 )
1798 else:
1799 if table is None:
1800 if self.table_fn:
1801 table = self.set_cls_attribute(
1802 "__table__", self.table_fn()
1803 )
1804 else:
1805 table = cls_as_Decl.__table__
1806 if declared_columns:
1807 for c in declared_columns:
1808 if not table.c.contains_column(c):
1809 raise exc.ArgumentError(
1810 "Can't add additional column %r when "
1811 "specifying __table__" % c.key
1812 )
1814 self.local_table = table
1816 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData:
1817 meta: Optional[MetaData] = getattr(self.cls, "metadata", None)
1818 if meta is not None:
1819 return meta
1820 else:
1821 return manager.registry.metadata
1823 def _setup_inheriting_mapper(self) -> None:
1824 cls = self.cls
1826 inherits = None
1828 if inherits is None:
1829 # since we search for classical mappings now, search for
1830 # multiple mapped bases as well and raise an error.
1831 inherits_search = []
1832 for base_ in cls.__bases__:
1833 c = _resolve_for_abstract_or_classical(base_)
1834 if c is None:
1835 continue
1837 if _is_supercls_for_inherits(c) and c not in inherits_search:
1838 inherits_search.append(c)
1840 if inherits_search:
1841 if len(inherits_search) > 1:
1842 raise exc.InvalidRequestError(
1843 "Class %s has multiple mapped bases: %r"
1844 % (cls, inherits_search)
1845 )
1846 inherits = inherits_search[0]
1847 elif isinstance(inherits, Mapper):
1848 inherits = inherits.class_
1850 self.inherits = inherits
1852 clsdict_view = self.clsdict_view
1853 if "__table__" not in clsdict_view and self.tablename is None:
1854 self.single = True
1856 def _setup_inheriting_columns(self) -> None:
1857 table = self.local_table
1858 cls = self.cls
1859 table_args = self.table_args
1860 declared_columns = self.declared_columns
1862 if (
1863 table is None
1864 and self.inherits is None
1865 and not _get_immediate_cls_attr(cls, "__no_table__")
1866 ):
1867 raise exc.InvalidRequestError(
1868 "Class %r does not have a __table__ or __tablename__ "
1869 "specified and does not inherit from an existing "
1870 "table-mapped class." % cls
1871 )
1872 elif self.inherits:
1873 inherited_mapper_or_config = _declared_mapping_info(self.inherits)
1874 assert inherited_mapper_or_config is not None
1875 inherited_table = inherited_mapper_or_config.local_table
1876 inherited_persist_selectable = (
1877 inherited_mapper_or_config.persist_selectable
1878 )
1880 if table is None:
1881 # single table inheritance.
1882 # ensure no table args
1883 if table_args:
1884 raise exc.ArgumentError(
1885 "Can't place __table_args__ on an inherited class "
1886 "with no table."
1887 )
1889 # add any columns declared here to the inherited table.
1890 if declared_columns and not isinstance(inherited_table, Table):
1891 raise exc.ArgumentError(
1892 f"Can't declare columns on single-table-inherited "
1893 f"subclass {self.cls}; superclass {self.inherits} "
1894 "is not mapped to a Table"
1895 )
1897 for col in declared_columns:
1898 assert inherited_table is not None
1899 if col.name in inherited_table.c:
1900 if inherited_table.c[col.name] is col:
1901 continue
1902 raise exc.ArgumentError(
1903 f"Column '{col}' on class {cls.__name__} "
1904 f"conflicts with existing column "
1905 f"'{inherited_table.c[col.name]}'. If using "
1906 f"Declarative, consider using the "
1907 "use_existing_column parameter of mapped_column() "
1908 "to resolve conflicts."
1909 )
1910 if col.primary_key:
1911 raise exc.ArgumentError(
1912 "Can't place primary key columns on an inherited "
1913 "class with no table."
1914 )
1916 if TYPE_CHECKING:
1917 assert isinstance(inherited_table, Table)
1919 inherited_table.append_column(col)
1920 if (
1921 inherited_persist_selectable is not None
1922 and inherited_persist_selectable is not inherited_table
1923 ):
1924 inherited_persist_selectable._refresh_for_new_column(
1925 col
1926 )
1928 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None:
1929 properties = self.properties
1931 if self.mapper_args_fn:
1932 mapper_args = self.mapper_args_fn()
1933 else:
1934 mapper_args = {}
1936 if mapper_kw:
1937 mapper_args.update(mapper_kw)
1939 if "properties" in mapper_args:
1940 properties = dict(properties)
1941 properties.update(mapper_args["properties"])
1943 # make sure that column copies are used rather
1944 # than the original columns from any mixins
1945 for k in ("version_id_col", "polymorphic_on"):
1946 if k in mapper_args:
1947 v = mapper_args[k]
1948 mapper_args[k] = self.column_copies.get(v, v)
1950 if "primary_key" in mapper_args:
1951 mapper_args["primary_key"] = [
1952 self.column_copies.get(v, v)
1953 for v in util.to_list(mapper_args["primary_key"])
1954 ]
1956 if "inherits" in mapper_args:
1957 inherits_arg = mapper_args["inherits"]
1958 if isinstance(inherits_arg, Mapper):
1959 inherits_arg = inherits_arg.class_
1961 if inherits_arg is not self.inherits:
1962 raise exc.InvalidRequestError(
1963 "mapper inherits argument given for non-inheriting "
1964 "class %s" % (mapper_args["inherits"])
1965 )
1967 if self.inherits:
1968 mapper_args["inherits"] = self.inherits
1970 if self.inherits and not mapper_args.get("concrete", False):
1971 # note the superclass is expected to have a Mapper assigned and
1972 # not be a deferred config, as this is called within map()
1973 inherited_mapper = class_mapper(self.inherits, False)
1974 inherited_table = inherited_mapper.local_table
1976 # single or joined inheritance
1977 # exclude any cols on the inherited table which are
1978 # not mapped on the parent class, to avoid
1979 # mapping columns specific to sibling/nephew classes
1980 if "exclude_properties" not in mapper_args:
1981 mapper_args["exclude_properties"] = exclude_properties = {
1982 c.key
1983 for c in inherited_table.c
1984 if c not in inherited_mapper._columntoproperty
1985 }.union(inherited_mapper.exclude_properties or ())
1986 exclude_properties.difference_update(
1987 [c.key for c in self.declared_columns]
1988 )
1990 # look through columns in the current mapper that
1991 # are keyed to a propname different than the colname
1992 # (if names were the same, we'd have popped it out above,
1993 # in which case the mapper makes this combination).
1994 # See if the superclass has a similar column property.
1995 # If so, join them together.
1996 for k, col in list(properties.items()):
1997 if not isinstance(col, expression.ColumnElement):
1998 continue
1999 if k in inherited_mapper._props:
2000 p = inherited_mapper._props[k]
2001 if isinstance(p, ColumnProperty):
2002 # note here we place the subclass column
2003 # first. See [ticket:1892] for background.
2004 properties[k] = [col] + p.columns
2005 result_mapper_args = mapper_args.copy()
2006 result_mapper_args["properties"] = properties
2007 self.mapper_args = result_mapper_args
2009 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2010 self._prepare_mapper_arguments(mapper_kw)
2011 if hasattr(self.cls, "__mapper_cls__"):
2012 mapper_cls = cast(
2013 "Type[Mapper[Any]]",
2014 util.unbound_method_to_callable(
2015 self.cls.__mapper_cls__ # type: ignore
2016 ),
2017 )
2018 else:
2019 mapper_cls = Mapper
2021 return self.set_cls_attribute(
2022 "__mapper__",
2023 mapper_cls(self.cls, self.local_table, **self.mapper_args),
2024 )
2027class _UnmappedDataclassConfig(_ClassScanAbstractConfig):
2028 """Configurator that will produce an unmapped dataclass."""
2030 __slots__ = (
2031 "clsdict_view",
2032 "collected_attributes",
2033 "collected_annotations",
2034 "allow_dataclass_fields",
2035 "dataclass_setup_arguments",
2036 "is_dataclass_prior_to_mapping",
2037 "allow_unmapped_annotations",
2038 )
2040 def __init__(
2041 self,
2042 cls_: Type[_O],
2043 dict_: _ClassDict,
2044 ):
2045 super().__init__(cls_)
2046 self.clsdict_view = (
2047 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
2048 )
2049 self.dataclass_setup_arguments = getattr(
2050 self.cls, "_sa_apply_dc_transforms", None
2051 )
2053 self.is_dataclass_prior_to_mapping = dataclasses.is_dataclass(cls_)
2054 self.allow_dataclass_fields = False
2055 self.allow_unmapped_annotations = True
2056 self.collected_attributes = {}
2057 self.collected_annotations = {}
2059 self._scan_attributes()
2061 self._setup_dataclasses_transforms(
2062 enable_descriptor_defaults=False, revert=True
2063 )
2065 def _scan_attributes(self) -> None:
2066 cls = self.cls
2068 clsdict_view = self.clsdict_view
2069 collected_attributes = self.collected_attributes
2070 _include_dunders = self._include_dunders
2072 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
2074 local_attributes_for_class = self._cls_attr_resolver(cls)
2075 for (
2076 name,
2077 obj,
2078 annotation,
2079 is_dataclass_field,
2080 ) in local_attributes_for_class():
2081 if name in _include_dunders:
2082 continue
2083 elif is_dataclass_field and (
2084 name not in clsdict_view or clsdict_view[name] is not obj
2085 ):
2086 # here, we are definitely looking at the target class
2087 # and not a superclass. this is currently a
2088 # dataclass-only path. if the name is only
2089 # a dataclass field and isn't in local cls.__dict__,
2090 # put the object there.
2091 # assert that the dataclass-enabled resolver agrees
2092 # with what we are seeing
2094 assert not attribute_is_overridden(name, obj)
2096 if _is_declarative_props(obj):
2097 obj = obj.fget()
2099 collected_attributes[name] = obj
2100 self._collect_annotation(name, annotation, cls, False, obj)
2101 else:
2102 self._collect_annotation(name, annotation, cls, None, obj)
2103 if name in clsdict_view:
2104 collected_attributes[name] = obj
2107@util.preload_module("sqlalchemy.orm.decl_api")
2108def _as_dc_declaredattr(
2109 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str
2110) -> Any:
2111 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr.
2112 # we can't write it because field.metadata is immutable :( so we have
2113 # to go through extra trouble to compare these
2114 decl_api = util.preloaded.orm_decl_api
2115 obj = field_metadata[sa_dataclass_metadata_key]
2116 if callable(obj) and not isinstance(obj, decl_api.declared_attr):
2117 return decl_api.declared_attr(obj)
2118 else:
2119 return obj
2122class _DeferredDeclarativeConfig(_DeclarativeMapperConfig):
2123 """Configurator that extends _DeclarativeMapperConfig to add a
2124 "deferred" step, to allow extensions like AbstractConcreteBase,
2125 DeferredMapping to partially set up a mapping that is "prepared"
2126 when table metadata is ready.
2128 """
2130 _cls: weakref.ref[Type[Any]]
2132 is_deferred = True
2134 _configs: util.OrderedDict[
2135 weakref.ref[Type[Any]], _DeferredDeclarativeConfig
2136 ] = util.OrderedDict()
2138 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
2139 pass
2141 @property
2142 def cls(self) -> Type[Any]:
2143 return self._cls() # type: ignore
2145 @cls.setter
2146 def cls(self, class_: Type[Any]) -> None:
2147 self._cls = weakref.ref(class_, self._remove_config_cls)
2148 self._configs[self._cls] = self
2150 @classmethod
2151 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None:
2152 cls._configs.pop(ref, None)
2154 @classmethod
2155 def has_cls(cls, class_: Type[Any]) -> bool:
2156 # 2.6 fails on weakref if class_ is an old style class
2157 return isinstance(class_, type) and weakref.ref(class_) in cls._configs
2159 @classmethod
2160 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn:
2161 if hasattr(class_, "_sa_raise_deferred_config"):
2162 class_._sa_raise_deferred_config()
2164 raise orm_exc.UnmappedClassError(
2165 class_,
2166 msg=(
2167 f"Class {orm_exc._safe_cls_name(class_)} has a deferred "
2168 "mapping on it. It is not yet usable as a mapped class."
2169 ),
2170 )
2172 @classmethod
2173 def config_for_cls(cls, class_: Type[Any]) -> _DeferredDeclarativeConfig:
2174 return cls._configs[weakref.ref(class_)]
2176 @classmethod
2177 def classes_for_base(
2178 cls, base_cls: Type[Any], sort: bool = True
2179 ) -> List[_DeferredDeclarativeConfig]:
2180 classes_for_base = [
2181 m
2182 for m, cls_ in [(m, m.cls) for m in cls._configs.values()]
2183 if cls_ is not None and issubclass(cls_, base_cls)
2184 ]
2186 if not sort:
2187 return classes_for_base
2189 all_m_by_cls = {m.cls: m for m in classes_for_base}
2191 tuples: List[
2192 Tuple[_DeferredDeclarativeConfig, _DeferredDeclarativeConfig]
2193 ] = []
2194 for m_cls in all_m_by_cls:
2195 tuples.extend(
2196 (all_m_by_cls[base_cls], all_m_by_cls[m_cls])
2197 for base_cls in m_cls.__bases__
2198 if base_cls in all_m_by_cls
2199 )
2200 return list(topological.sort(tuples, classes_for_base))
2202 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2203 self._configs.pop(self._cls, None)
2204 return super().map(mapper_kw)
2207def _add_attribute(
2208 cls: Type[Any], key: str, value: MapperProperty[Any]
2209) -> None:
2210 """add an attribute to an existing declarative class.
2212 This runs through the logic to determine MapperProperty,
2213 adds it to the Mapper, adds a column to the mapped Table, etc.
2215 """
2217 if "__mapper__" in cls.__dict__:
2218 mapped_cls = cast("MappedClassProtocol[Any]", cls)
2220 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table:
2221 if isinstance(mc.__table__, Table):
2222 return mc.__table__
2223 raise exc.InvalidRequestError(
2224 f"Cannot add a new attribute to mapped class {mc.__name__!r} "
2225 "because it's not mapped against a table."
2226 )
2228 if isinstance(value, Column):
2229 _undefer_column_name(key, value)
2230 _table_or_raise(mapped_cls).append_column(
2231 value, replace_existing=True
2232 )
2233 mapped_cls.__mapper__.add_property(key, value)
2234 elif isinstance(value, _MapsColumns):
2235 mp = value.mapper_property_to_assign
2236 for col, _ in value.columns_to_assign:
2237 _undefer_column_name(key, col)
2238 _table_or_raise(mapped_cls).append_column(
2239 col, replace_existing=True
2240 )
2241 if not mp:
2242 mapped_cls.__mapper__.add_property(key, col)
2243 if mp:
2244 mapped_cls.__mapper__.add_property(key, mp)
2245 elif isinstance(value, MapperProperty):
2246 mapped_cls.__mapper__.add_property(key, value)
2247 elif isinstance(value, QueryableAttribute) and value.key != key:
2248 # detect a QueryableAttribute that's already mapped being
2249 # assigned elsewhere in userland, turn into a synonym()
2250 value = SynonymProperty(value.key)
2251 mapped_cls.__mapper__.add_property(key, value)
2252 else:
2253 type.__setattr__(cls, key, value)
2254 mapped_cls.__mapper__._expire_memoizations()
2255 else:
2256 type.__setattr__(cls, key, value)
2259def _del_attribute(cls: Type[Any], key: str) -> None:
2260 if (
2261 "__mapper__" in cls.__dict__
2262 and key in cls.__dict__
2263 and not cast(
2264 "MappedClassProtocol[Any]", cls
2265 ).__mapper__._dispose_called
2266 ):
2267 value = cls.__dict__[key]
2268 if isinstance(
2269 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute)
2270 ):
2271 raise NotImplementedError(
2272 "Can't un-map individual mapped attributes on a mapped class."
2273 )
2274 else:
2275 type.__delattr__(cls, key)
2276 cast(
2277 "MappedClassProtocol[Any]", cls
2278 ).__mapper__._expire_memoizations()
2279 else:
2280 type.__delattr__(cls, key)
2283def _declarative_constructor(self: Any, **kwargs: Any) -> None:
2284 """A simple constructor that allows initialization from kwargs.
2286 Sets attributes on the constructed instance using the names and
2287 values in ``kwargs``.
2289 Only keys that are present as
2290 attributes of the instance's class are allowed. These could be,
2291 for example, any mapped columns or relationships.
2292 """
2293 cls_ = type(self)
2294 for k in kwargs:
2295 if not hasattr(cls_, k):
2296 raise TypeError(
2297 "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
2298 )
2299 setattr(self, k, kwargs[k])
2302_declarative_constructor.__name__ = "__init__"
2305def _undefer_column_name(key: str, column: Column[Any]) -> None:
2306 if column.key is None:
2307 column.key = key
2308 if column.name is None:
2309 column.name = key