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