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 re
15from typing import Any
16from typing import Callable
17from typing import cast
18from typing import Dict
19from typing import Iterable
20from typing import List
21from typing import Mapping
22from typing import NamedTuple
23from typing import NoReturn
24from typing import Optional
25from typing import Sequence
26from typing import Tuple
27from typing import Type
28from typing import TYPE_CHECKING
29from typing import TypeVar
30from typing import Union
31import weakref
33from . import attributes
34from . import clsregistry
35from . import exc as orm_exc
36from . import instrumentation
37from . import mapperlib
38from ._typing import _O
39from ._typing import attr_is_internal_proxy
40from .attributes import InstrumentedAttribute
41from .attributes import QueryableAttribute
42from .base import _is_mapped_class
43from .base import InspectionAttr
44from .descriptor_props import CompositeProperty
45from .descriptor_props import SynonymProperty
46from .interfaces import _AttributeOptions
47from .interfaces import _DCAttributeOptions
48from .interfaces import _IntrospectsAnnotations
49from .interfaces import _MappedAttribute
50from .interfaces import _MapsColumns
51from .interfaces import MapperProperty
52from .mapper import Mapper
53from .properties import ColumnProperty
54from .properties import MappedColumn
55from .util import _extract_mapped_subtype
56from .util import _is_mapped_annotation
57from .util import class_mapper
58from .util import de_stringify_annotation
59from .. import event
60from .. import exc
61from .. import util
62from ..sql import expression
63from ..sql.base import _NoArg
64from ..sql.schema import Column
65from ..sql.schema import Table
66from ..util import topological
67from ..util.typing import _AnnotationScanType
68from ..util.typing import get_args
69from ..util.typing import is_fwd_ref
70from ..util.typing import is_literal
71from ..util.typing import Protocol
72from ..util.typing import TypedDict
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: ...
119class _DataclassArguments(TypedDict):
120 init: Union[_NoArg, bool]
121 repr: Union[_NoArg, bool]
122 eq: Union[_NoArg, bool]
123 order: Union[_NoArg, bool]
124 unsafe_hash: Union[_NoArg, bool]
125 match_args: Union[_NoArg, bool]
126 kw_only: Union[_NoArg, bool]
127 dataclass_callable: Union[_NoArg, Callable[..., Type[Any]]]
130def _declared_mapping_info(
131 cls: Type[Any],
132) -> Optional[Union[_DeferredMapperConfig, Mapper[Any]]]:
133 # deferred mapping
134 if _DeferredMapperConfig.has_cls(cls):
135 return _DeferredMapperConfig.config_for_cls(cls)
136 # regular mapping
137 elif _is_mapped_class(cls):
138 return class_mapper(cls, configure=False)
139 else:
140 return None
143def _is_supercls_for_inherits(cls: Type[Any]) -> bool:
144 """return True if this class will be used as a superclass to set in
145 'inherits'.
147 This includes deferred mapper configs that aren't mapped yet, however does
148 not include classes with _sa_decl_prepare_nocascade (e.g.
149 ``AbstractConcreteBase``); these concrete-only classes are not set up as
150 "inherits" until after mappers are configured using
151 mapper._set_concrete_base()
153 """
154 if _DeferredMapperConfig.has_cls(cls):
155 return not _get_immediate_cls_attr(
156 cls, "_sa_decl_prepare_nocascade", strict=True
157 )
158 # regular mapping
159 elif _is_mapped_class(cls):
160 return True
161 else:
162 return False
165def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
166 if cls is object:
167 return None
169 sup: Optional[Type[Any]]
171 if cls.__dict__.get("__abstract__", False):
172 for base_ in cls.__bases__:
173 sup = _resolve_for_abstract_or_classical(base_)
174 if sup is not None:
175 return sup
176 else:
177 return None
178 else:
179 clsmanager = _dive_for_cls_manager(cls)
181 if clsmanager:
182 return clsmanager.class_
183 else:
184 return cls
187def _get_immediate_cls_attr(
188 cls: Type[Any], attrname: str, strict: bool = False
189) -> Optional[Any]:
190 """return an attribute of the class that is either present directly
191 on the class, e.g. not on a superclass, or is from a superclass but
192 this superclass is a non-mapped mixin, that is, not a descendant of
193 the declarative base and is also not classically mapped.
195 This is used to detect attributes that indicate something about
196 a mapped class independently from any mapped classes that it may
197 inherit from.
199 """
201 # the rules are different for this name than others,
202 # make sure we've moved it out. transitional
203 assert attrname != "__abstract__"
205 if not issubclass(cls, object):
206 return None
208 if attrname in cls.__dict__:
209 return getattr(cls, attrname)
211 for base in cls.__mro__[1:]:
212 _is_classical_inherits = _dive_for_cls_manager(base) is not None
214 if attrname in base.__dict__ and (
215 base is cls
216 or (
217 (base in cls.__bases__ if strict else True)
218 and not _is_classical_inherits
219 )
220 ):
221 return getattr(base, attrname)
222 else:
223 return None
226def _dive_for_cls_manager(cls: Type[_O]) -> Optional[ClassManager[_O]]:
227 # because the class manager registration is pluggable,
228 # we need to do the search for every class in the hierarchy,
229 # rather than just a simple "cls._sa_class_manager"
231 for base in cls.__mro__:
232 manager: Optional[ClassManager[_O]] = attributes.opt_manager_of_class(
233 base
234 )
235 if manager:
236 return manager
237 return None
240def _as_declarative(
241 registry: _RegistryType, cls: Type[Any], dict_: _ClassDict
242) -> Optional[_MapperConfig]:
243 # declarative scans the class for attributes. no table or mapper
244 # args passed separately.
245 return _MapperConfig.setup_mapping(registry, cls, dict_, None, {})
248def _mapper(
249 registry: _RegistryType,
250 cls: Type[_O],
251 table: Optional[FromClause],
252 mapper_kw: _MapperKwArgs,
253) -> Mapper[_O]:
254 _ImperativeMapperConfig(registry, cls, table, mapper_kw)
255 return cast("MappedClassProtocol[_O]", cls).__mapper__
258@util.preload_module("sqlalchemy.orm.decl_api")
259def _is_declarative_props(obj: Any) -> bool:
260 _declared_attr_common = util.preloaded.orm_decl_api._declared_attr_common
262 return isinstance(obj, (_declared_attr_common, util.classproperty))
265def _check_declared_props_nocascade(
266 obj: Any, name: str, cls: Type[_O]
267) -> bool:
268 if _is_declarative_props(obj):
269 if getattr(obj, "_cascading", False):
270 util.warn(
271 "@declared_attr.cascading is not supported on the %s "
272 "attribute on class %s. This attribute invokes for "
273 "subclasses in any case." % (name, cls)
274 )
275 return True
276 else:
277 return False
280class _MapperConfig:
281 __slots__ = (
282 "cls",
283 "classname",
284 "properties",
285 "declared_attr_reg",
286 "__weakref__",
287 )
289 cls: Type[Any]
290 classname: str
291 properties: util.OrderedDict[
292 str,
293 Union[
294 Sequence[NamedColumn[Any]], NamedColumn[Any], MapperProperty[Any]
295 ],
296 ]
297 declared_attr_reg: Dict[declared_attr[Any], Any]
299 @classmethod
300 def setup_mapping(
301 cls,
302 registry: _RegistryType,
303 cls_: Type[_O],
304 dict_: _ClassDict,
305 table: Optional[FromClause],
306 mapper_kw: _MapperKwArgs,
307 ) -> Optional[_MapperConfig]:
308 manager = attributes.opt_manager_of_class(cls)
309 if manager and manager.class_ is cls_:
310 raise exc.InvalidRequestError(
311 f"Class {cls!r} already has been instrumented declaratively"
312 )
314 if cls_.__dict__.get("__abstract__", False):
315 return None
317 defer_map = _get_immediate_cls_attr(
318 cls_, "_sa_decl_prepare_nocascade", strict=True
319 ) or hasattr(cls_, "_sa_decl_prepare")
321 if defer_map:
322 return _DeferredMapperConfig(
323 registry, cls_, dict_, table, mapper_kw
324 )
325 else:
326 return _ClassScanMapperConfig(
327 registry, cls_, dict_, table, mapper_kw
328 )
330 def __init__(
331 self,
332 registry: _RegistryType,
333 cls_: Type[Any],
334 mapper_kw: _MapperKwArgs,
335 ):
336 self.cls = util.assert_arg_type(cls_, type, "cls_")
337 self.classname = cls_.__name__
338 self.properties = util.OrderedDict()
339 self.declared_attr_reg = {}
341 if not mapper_kw.get("non_primary", False):
342 instrumentation.register_class(
343 self.cls,
344 finalize=False,
345 registry=registry,
346 declarative_scan=self,
347 init_method=registry.constructor,
348 )
349 else:
350 manager = attributes.opt_manager_of_class(self.cls)
351 if not manager or not manager.is_mapped:
352 raise exc.InvalidRequestError(
353 "Class %s has no primary mapper configured. Configure "
354 "a primary mapper first before setting up a non primary "
355 "Mapper." % self.cls
356 )
358 def set_cls_attribute(self, attrname: str, value: _T) -> _T:
359 manager = instrumentation.manager_of_class(self.cls)
360 manager.install_member(attrname, value)
361 return value
363 def map(self, mapper_kw: _MapperKwArgs = ...) -> Mapper[Any]:
364 raise NotImplementedError()
366 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
367 self.map(mapper_kw)
370class _ImperativeMapperConfig(_MapperConfig):
371 __slots__ = ("local_table", "inherits")
373 def __init__(
374 self,
375 registry: _RegistryType,
376 cls_: Type[_O],
377 table: Optional[FromClause],
378 mapper_kw: _MapperKwArgs,
379 ):
380 super().__init__(registry, cls_, mapper_kw)
382 self.local_table = self.set_cls_attribute("__table__", table)
384 with mapperlib._CONFIGURE_MUTEX:
385 if not mapper_kw.get("non_primary", False):
386 clsregistry.add_class(
387 self.classname, self.cls, registry._class_registry
388 )
390 self._setup_inheritance(mapper_kw)
392 self._early_mapping(mapper_kw)
394 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
395 mapper_cls = Mapper
397 return self.set_cls_attribute(
398 "__mapper__",
399 mapper_cls(self.cls, self.local_table, **mapper_kw),
400 )
402 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
403 cls = self.cls
405 inherits = mapper_kw.get("inherits", None)
407 if inherits is None:
408 # since we search for classical mappings now, search for
409 # multiple mapped bases as well and raise an error.
410 inherits_search = []
411 for base_ in cls.__bases__:
412 c = _resolve_for_abstract_or_classical(base_)
413 if c is None:
414 continue
416 if _is_supercls_for_inherits(c) and c not in inherits_search:
417 inherits_search.append(c)
419 if inherits_search:
420 if len(inherits_search) > 1:
421 raise exc.InvalidRequestError(
422 "Class %s has multiple mapped bases: %r"
423 % (cls, inherits_search)
424 )
425 inherits = inherits_search[0]
426 elif isinstance(inherits, Mapper):
427 inherits = inherits.class_
429 self.inherits = inherits
432class _CollectedAnnotation(NamedTuple):
433 raw_annotation: _AnnotationScanType
434 mapped_container: Optional[Type[Mapped[Any]]]
435 extracted_mapped_annotation: Union[_AnnotationScanType, str]
436 is_dataclass: bool
437 attr_value: Any
438 originating_module: str
439 originating_class: Type[Any]
442class _ClassScanMapperConfig(_MapperConfig):
443 __slots__ = (
444 "registry",
445 "clsdict_view",
446 "collected_attributes",
447 "collected_annotations",
448 "local_table",
449 "persist_selectable",
450 "declared_columns",
451 "column_ordering",
452 "column_copies",
453 "table_args",
454 "tablename",
455 "mapper_args",
456 "mapper_args_fn",
457 "table_fn",
458 "inherits",
459 "single",
460 "allow_dataclass_fields",
461 "dataclass_setup_arguments",
462 "is_dataclass_prior_to_mapping",
463 "allow_unmapped_annotations",
464 )
466 is_deferred = False
467 registry: _RegistryType
468 clsdict_view: _ClassDict
469 collected_annotations: Dict[str, _CollectedAnnotation]
470 collected_attributes: Dict[str, Any]
471 local_table: Optional[FromClause]
472 persist_selectable: Optional[FromClause]
473 declared_columns: util.OrderedSet[Column[Any]]
474 column_ordering: Dict[Column[Any], int]
475 column_copies: Dict[
476 Union[MappedColumn[Any], Column[Any]],
477 Union[MappedColumn[Any], Column[Any]],
478 ]
479 tablename: Optional[str]
480 mapper_args: Mapping[str, Any]
481 table_args: Optional[_TableArgsType]
482 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
483 inherits: Optional[Type[Any]]
484 single: bool
486 is_dataclass_prior_to_mapping: bool
487 allow_unmapped_annotations: bool
489 dataclass_setup_arguments: Optional[_DataclassArguments]
490 """if the class has SQLAlchemy native dataclass parameters, where
491 we will turn the class into a dataclass within the declarative mapping
492 process.
494 """
496 allow_dataclass_fields: bool
497 """if true, look for dataclass-processed Field objects on the target
498 class as well as superclasses and extract ORM mapping directives from
499 the "metadata" attribute of each Field.
501 if False, dataclass fields can still be used, however they won't be
502 mapped.
504 """
506 def __init__(
507 self,
508 registry: _RegistryType,
509 cls_: Type[_O],
510 dict_: _ClassDict,
511 table: Optional[FromClause],
512 mapper_kw: _MapperKwArgs,
513 ):
514 # grab class dict before the instrumentation manager has been added.
515 # reduces cycles
516 self.clsdict_view = (
517 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
518 )
519 super().__init__(registry, cls_, mapper_kw)
520 self.registry = registry
521 self.persist_selectable = None
523 self.collected_attributes = {}
524 self.collected_annotations = {}
525 self.declared_columns = util.OrderedSet()
526 self.column_ordering = {}
527 self.column_copies = {}
528 self.single = False
529 self.dataclass_setup_arguments = dca = getattr(
530 self.cls, "_sa_apply_dc_transforms", None
531 )
533 self.allow_unmapped_annotations = getattr(
534 self.cls, "__allow_unmapped__", False
535 ) or bool(self.dataclass_setup_arguments)
537 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
538 cls_
539 )
541 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
543 # we don't want to consume Field objects from a not-already-dataclass.
544 # the Field objects won't have their "name" or "type" populated,
545 # and while it seems like we could just set these on Field as we
546 # read them, Field is documented as "user read only" and we need to
547 # stay far away from any off-label use of dataclasses APIs.
548 if (not cld or dca) and sdk:
549 raise exc.InvalidRequestError(
550 "SQLAlchemy mapped dataclasses can't consume mapping "
551 "information from dataclass.Field() objects if the immediate "
552 "class is not already a dataclass."
553 )
555 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
556 # then also look inside of dataclass.Field() objects yielded by
557 # dataclasses.get_fields(cls) when scanning for attributes
558 self.allow_dataclass_fields = bool(sdk and cld)
560 self._setup_declared_events()
562 self._scan_attributes()
564 self._setup_dataclasses_transforms()
566 with mapperlib._CONFIGURE_MUTEX:
567 clsregistry.add_class(
568 self.classname, self.cls, registry._class_registry
569 )
571 self._setup_inheriting_mapper(mapper_kw)
573 self._extract_mappable_attributes()
575 self._extract_declared_columns()
577 self._setup_table(table)
579 self._setup_inheriting_columns(mapper_kw)
581 self._early_mapping(mapper_kw)
583 def _setup_declared_events(self) -> None:
584 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
586 @event.listens_for(Mapper, "after_configured")
587 def after_configured() -> None:
588 cast(
589 "_DeclMappedClassProtocol[Any]", self.cls
590 ).__declare_last__()
592 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
594 @event.listens_for(Mapper, "before_configured")
595 def before_configured() -> None:
596 cast(
597 "_DeclMappedClassProtocol[Any]", self.cls
598 ).__declare_first__()
600 def _cls_attr_override_checker(
601 self, cls: Type[_O]
602 ) -> Callable[[str, Any], bool]:
603 """Produce a function that checks if a class has overridden an
604 attribute, taking SQLAlchemy-enabled dataclass fields into account.
606 """
608 if self.allow_dataclass_fields:
609 sa_dataclass_metadata_key = _get_immediate_cls_attr(
610 cls, "__sa_dataclass_metadata_key__"
611 )
612 else:
613 sa_dataclass_metadata_key = None
615 if not sa_dataclass_metadata_key:
617 def attribute_is_overridden(key: str, obj: Any) -> bool:
618 return getattr(cls, key, obj) is not obj
620 else:
621 all_datacls_fields = {
622 f.name: f.metadata[sa_dataclass_metadata_key]
623 for f in util.dataclass_fields(cls)
624 if sa_dataclass_metadata_key in f.metadata
625 }
626 local_datacls_fields = {
627 f.name: f.metadata[sa_dataclass_metadata_key]
628 for f in util.local_dataclass_fields(cls)
629 if sa_dataclass_metadata_key in f.metadata
630 }
632 absent = object()
634 def attribute_is_overridden(key: str, obj: Any) -> bool:
635 if _is_declarative_props(obj):
636 obj = obj.fget
638 # this function likely has some failure modes still if
639 # someone is doing a deep mixing of the same attribute
640 # name as plain Python attribute vs. dataclass field.
642 ret = local_datacls_fields.get(key, absent)
643 if _is_declarative_props(ret):
644 ret = ret.fget
646 if ret is obj:
647 return False
648 elif ret is not absent:
649 return True
651 all_field = all_datacls_fields.get(key, absent)
653 ret = getattr(cls, key, obj)
655 if ret is obj:
656 return False
658 # for dataclasses, this could be the
659 # 'default' of the field. so filter more specifically
660 # for an already-mapped InstrumentedAttribute
661 if ret is not absent and isinstance(
662 ret, InstrumentedAttribute
663 ):
664 return True
666 if all_field is obj:
667 return False
668 elif all_field is not absent:
669 return True
671 # can't find another attribute
672 return False
674 return attribute_is_overridden
676 _include_dunders = {
677 "__table__",
678 "__mapper_args__",
679 "__tablename__",
680 "__table_args__",
681 }
683 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
685 def _cls_attr_resolver(
686 self, cls: Type[Any]
687 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
688 """produce a function to iterate the "attributes" of a class
689 which we want to consider for mapping, adjusting for SQLAlchemy fields
690 embedded in dataclass fields.
692 """
693 cls_annotations = util.get_annotations(cls)
695 cls_vars = vars(cls)
697 _include_dunders = self._include_dunders
698 _match_exclude_dunders = self._match_exclude_dunders
700 names = [
701 n
702 for n in util.merge_lists_w_ordering(
703 list(cls_vars), list(cls_annotations)
704 )
705 if not _match_exclude_dunders.match(n) or n in _include_dunders
706 ]
708 if self.allow_dataclass_fields:
709 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
710 cls, "__sa_dataclass_metadata_key__"
711 )
712 else:
713 sa_dataclass_metadata_key = None
715 if not sa_dataclass_metadata_key:
717 def local_attributes_for_class() -> (
718 Iterable[Tuple[str, Any, Any, bool]]
719 ):
720 return (
721 (
722 name,
723 cls_vars.get(name),
724 cls_annotations.get(name),
725 False,
726 )
727 for name in names
728 )
730 else:
731 dataclass_fields = {
732 field.name: field for field in util.local_dataclass_fields(cls)
733 }
735 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
737 def local_attributes_for_class() -> (
738 Iterable[Tuple[str, Any, Any, bool]]
739 ):
740 for name in names:
741 field = dataclass_fields.get(name, None)
742 if field and sa_dataclass_metadata_key in field.metadata:
743 yield field.name, _as_dc_declaredattr(
744 field.metadata, fixed_sa_dataclass_metadata_key
745 ), cls_annotations.get(field.name), True
746 else:
747 yield name, cls_vars.get(name), cls_annotations.get(
748 name
749 ), False
751 return local_attributes_for_class
753 def _scan_attributes(self) -> None:
754 cls = self.cls
756 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
758 clsdict_view = self.clsdict_view
759 collected_attributes = self.collected_attributes
760 column_copies = self.column_copies
761 _include_dunders = self._include_dunders
762 mapper_args_fn = None
763 table_args = inherited_table_args = None
764 table_fn = None
765 tablename = None
766 fixed_table = "__table__" in clsdict_view
768 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
770 bases = []
772 for base in cls.__mro__:
773 # collect bases and make sure standalone columns are copied
774 # to be the column they will ultimately be on the class,
775 # so that declared_attr functions use the right columns.
776 # need to do this all the way up the hierarchy first
777 # (see #8190)
779 class_mapped = base is not cls and _is_supercls_for_inherits(base)
781 local_attributes_for_class = self._cls_attr_resolver(base)
783 if not class_mapped and base is not cls:
784 locally_collected_columns = self._produce_column_copies(
785 local_attributes_for_class,
786 attribute_is_overridden,
787 fixed_table,
788 base,
789 )
790 else:
791 locally_collected_columns = {}
793 bases.append(
794 (
795 base,
796 class_mapped,
797 local_attributes_for_class,
798 locally_collected_columns,
799 )
800 )
802 for (
803 base,
804 class_mapped,
805 local_attributes_for_class,
806 locally_collected_columns,
807 ) in bases:
808 # this transfer can also take place as we scan each name
809 # for finer-grained control of how collected_attributes is
810 # populated, as this is what impacts column ordering.
811 # however it's simpler to get it out of the way here.
812 collected_attributes.update(locally_collected_columns)
814 for (
815 name,
816 obj,
817 annotation,
818 is_dataclass_field,
819 ) in local_attributes_for_class():
820 if name in _include_dunders:
821 if name == "__mapper_args__":
822 check_decl = _check_declared_props_nocascade(
823 obj, name, cls
824 )
825 if not mapper_args_fn and (
826 not class_mapped or check_decl
827 ):
828 # don't even invoke __mapper_args__ until
829 # after we've determined everything about the
830 # mapped table.
831 # make a copy of it so a class-level dictionary
832 # is not overwritten when we update column-based
833 # arguments.
834 def _mapper_args_fn() -> Dict[str, Any]:
835 return dict(cls_as_Decl.__mapper_args__)
837 mapper_args_fn = _mapper_args_fn
839 elif name == "__tablename__":
840 check_decl = _check_declared_props_nocascade(
841 obj, name, cls
842 )
843 if not tablename and (not class_mapped or check_decl):
844 tablename = cls_as_Decl.__tablename__
845 elif name == "__table__":
846 check_decl = _check_declared_props_nocascade(
847 obj, name, cls
848 )
849 # if a @declared_attr using "__table__" is detected,
850 # wrap up a callable to look for "__table__" from
851 # the final concrete class when we set up a table.
852 # this was fixed by
853 # #11509, regression in 2.0 from version 1.4.
854 if check_decl and not table_fn:
855 # don't even invoke __table__ until we're ready
856 def _table_fn() -> FromClause:
857 return cls_as_Decl.__table__
859 table_fn = _table_fn
861 elif name == "__table_args__":
862 check_decl = _check_declared_props_nocascade(
863 obj, name, cls
864 )
865 if not table_args and (not class_mapped or check_decl):
866 table_args = cls_as_Decl.__table_args__
867 if not isinstance(
868 table_args, (tuple, dict, type(None))
869 ):
870 raise exc.ArgumentError(
871 "__table_args__ value must be a tuple, "
872 "dict, or None"
873 )
874 if base is not cls:
875 inherited_table_args = True
876 else:
877 # any other dunder names; should not be here
878 # as we have tested for all four names in
879 # _include_dunders
880 assert False
881 elif class_mapped:
882 if _is_declarative_props(obj) and not obj._quiet:
883 util.warn(
884 "Regular (i.e. not __special__) "
885 "attribute '%s.%s' uses @declared_attr, "
886 "but owning class %s is mapped - "
887 "not applying to subclass %s."
888 % (base.__name__, name, base, cls)
889 )
891 continue
892 elif base is not cls:
893 # we're a mixin, abstract base, or something that is
894 # acting like that for now.
896 if isinstance(obj, (Column, MappedColumn)):
897 # already copied columns to the mapped class.
898 continue
899 elif isinstance(obj, MapperProperty):
900 raise exc.InvalidRequestError(
901 "Mapper properties (i.e. deferred,"
902 "column_property(), relationship(), etc.) must "
903 "be declared as @declared_attr callables "
904 "on declarative mixin classes. For dataclass "
905 "field() objects, use a lambda:"
906 )
907 elif _is_declarative_props(obj):
908 # tried to get overloads to tell this to
909 # pylance, no luck
910 assert obj is not None
912 if obj._cascading:
913 if name in clsdict_view:
914 # unfortunately, while we can use the user-
915 # defined attribute here to allow a clean
916 # override, if there's another
917 # subclass below then it still tries to use
918 # this. not sure if there is enough
919 # information here to add this as a feature
920 # later on.
921 util.warn(
922 "Attribute '%s' on class %s cannot be "
923 "processed due to "
924 "@declared_attr.cascading; "
925 "skipping" % (name, cls)
926 )
927 collected_attributes[name] = column_copies[obj] = (
928 ret
929 ) = obj.__get__(obj, cls)
930 setattr(cls, name, ret)
931 else:
932 if is_dataclass_field:
933 # access attribute using normal class access
934 # first, to see if it's been mapped on a
935 # superclass. note if the dataclasses.field()
936 # has "default", this value can be anything.
937 ret = getattr(cls, name, None)
939 # so, if it's anything that's not ORM
940 # mapped, assume we should invoke the
941 # declared_attr
942 if not isinstance(ret, InspectionAttr):
943 ret = obj.fget()
944 else:
945 # access attribute using normal class access.
946 # if the declared attr already took place
947 # on a superclass that is mapped, then
948 # this is no longer a declared_attr, it will
949 # be the InstrumentedAttribute
950 ret = getattr(cls, name)
952 # correct for proxies created from hybrid_property
953 # or similar. note there is no known case that
954 # produces nested proxies, so we are only
955 # looking one level deep right now.
957 if (
958 isinstance(ret, InspectionAttr)
959 and attr_is_internal_proxy(ret)
960 and not isinstance(
961 ret.original_property, MapperProperty
962 )
963 ):
964 ret = ret.descriptor
966 collected_attributes[name] = column_copies[obj] = (
967 ret
968 )
970 if (
971 isinstance(ret, (Column, MapperProperty))
972 and ret.doc is None
973 ):
974 ret.doc = obj.__doc__
976 self._collect_annotation(
977 name,
978 obj._collect_return_annotation(),
979 base,
980 True,
981 obj,
982 )
983 elif _is_mapped_annotation(annotation, cls, base):
984 # Mapped annotation without any object.
985 # product_column_copies should have handled this.
986 # if future support for other MapperProperty,
987 # then test if this name is already handled and
988 # otherwise proceed to generate.
989 if not fixed_table:
990 assert (
991 name in collected_attributes
992 or attribute_is_overridden(name, None)
993 )
994 continue
995 else:
996 # here, the attribute is some other kind of
997 # property that we assume is not part of the
998 # declarative mapping. however, check for some
999 # more common mistakes
1000 self._warn_for_decl_attributes(base, name, obj)
1001 elif is_dataclass_field and (
1002 name not in clsdict_view or clsdict_view[name] is not obj
1003 ):
1004 # here, we are definitely looking at the target class
1005 # and not a superclass. this is currently a
1006 # dataclass-only path. if the name is only
1007 # a dataclass field and isn't in local cls.__dict__,
1008 # put the object there.
1009 # assert that the dataclass-enabled resolver agrees
1010 # with what we are seeing
1012 assert not attribute_is_overridden(name, obj)
1014 if _is_declarative_props(obj):
1015 obj = obj.fget()
1017 collected_attributes[name] = obj
1018 self._collect_annotation(
1019 name, annotation, base, False, obj
1020 )
1021 else:
1022 collected_annotation = self._collect_annotation(
1023 name, annotation, base, None, obj
1024 )
1025 is_mapped = (
1026 collected_annotation is not None
1027 and collected_annotation.mapped_container is not None
1028 )
1029 generated_obj = (
1030 collected_annotation.attr_value
1031 if collected_annotation is not None
1032 else obj
1033 )
1034 if obj is None and not fixed_table and is_mapped:
1035 collected_attributes[name] = (
1036 generated_obj
1037 if generated_obj is not None
1038 else MappedColumn()
1039 )
1040 elif name in clsdict_view:
1041 collected_attributes[name] = obj
1042 # else if the name is not in the cls.__dict__,
1043 # don't collect it as an attribute.
1044 # we will see the annotation only, which is meaningful
1045 # both for mapping and dataclasses setup
1047 if inherited_table_args and not tablename:
1048 table_args = None
1050 self.table_args = table_args
1051 self.tablename = tablename
1052 self.mapper_args_fn = mapper_args_fn
1053 self.table_fn = table_fn
1055 def _setup_dataclasses_transforms(self) -> None:
1056 dataclass_setup_arguments = self.dataclass_setup_arguments
1057 if not dataclass_setup_arguments:
1058 return
1060 # can't use is_dataclass since it uses hasattr
1061 if "__dataclass_fields__" in self.cls.__dict__:
1062 raise exc.InvalidRequestError(
1063 f"Class {self.cls} is already a dataclass; ensure that "
1064 "base classes / decorator styles of establishing dataclasses "
1065 "are not being mixed. "
1066 "This can happen if a class that inherits from "
1067 "'MappedAsDataclass', even indirectly, is been mapped with "
1068 "'@registry.mapped_as_dataclass'"
1069 )
1071 # can't create a dataclass if __table__ is already there. This would
1072 # fail an assertion when calling _get_arguments_for_make_dataclass:
1073 # assert False, "Mapped[] received without a mapping declaration"
1074 if "__table__" in self.cls.__dict__:
1075 raise exc.InvalidRequestError(
1076 f"Class {self.cls} already defines a '__table__'. "
1077 "ORM Annotated Dataclasses do not support a pre-existing "
1078 "'__table__' element"
1079 )
1081 warn_for_non_dc_attrs = collections.defaultdict(list)
1083 def _allow_dataclass_field(
1084 key: str, originating_class: Type[Any]
1085 ) -> bool:
1086 if (
1087 originating_class is not self.cls
1088 and "__dataclass_fields__" not in originating_class.__dict__
1089 ):
1090 warn_for_non_dc_attrs[originating_class].append(key)
1092 return True
1094 manager = instrumentation.manager_of_class(self.cls)
1095 assert manager is not None
1097 field_list = [
1098 _AttributeOptions._get_arguments_for_make_dataclass(
1099 key,
1100 anno,
1101 mapped_container,
1102 self.collected_attributes.get(key, _NoArg.NO_ARG),
1103 )
1104 for key, anno, mapped_container in (
1105 (
1106 key,
1107 mapped_anno if mapped_anno else raw_anno,
1108 mapped_container,
1109 )
1110 for key, (
1111 raw_anno,
1112 mapped_container,
1113 mapped_anno,
1114 is_dc,
1115 attr_value,
1116 originating_module,
1117 originating_class,
1118 ) in self.collected_annotations.items()
1119 if _allow_dataclass_field(key, originating_class)
1120 and (
1121 key not in self.collected_attributes
1122 # issue #9226; check for attributes that we've collected
1123 # which are already instrumented, which we would assume
1124 # mean we are in an ORM inheritance mapping and this
1125 # attribute is already mapped on the superclass. Under
1126 # no circumstance should any QueryableAttribute be sent to
1127 # the dataclass() function; anything that's mapped should
1128 # be Field and that's it
1129 or not isinstance(
1130 self.collected_attributes[key], QueryableAttribute
1131 )
1132 )
1133 )
1134 ]
1136 if warn_for_non_dc_attrs:
1137 for (
1138 originating_class,
1139 non_dc_attrs,
1140 ) in warn_for_non_dc_attrs.items():
1141 util.warn_deprecated(
1142 f"When transforming {self.cls} to a dataclass, "
1143 f"attribute(s) "
1144 f"{', '.join(repr(key) for key in non_dc_attrs)} "
1145 f"originates from superclass "
1146 f"{originating_class}, which is not a dataclass. This "
1147 f"usage is deprecated and will raise an error in "
1148 f"SQLAlchemy 2.1. When declaring SQLAlchemy Declarative "
1149 f"Dataclasses, ensure that all mixin classes and other "
1150 f"superclasses which include attributes are also a "
1151 f"subclass of MappedAsDataclass.",
1152 "2.0",
1153 code="dcmx",
1154 )
1156 annotations = {}
1157 defaults = {}
1158 for item in field_list:
1159 if len(item) == 2:
1160 name, tp = item
1161 elif len(item) == 3:
1162 name, tp, spec = item
1163 defaults[name] = spec
1164 else:
1165 assert False
1166 annotations[name] = tp
1168 for k, v in defaults.items():
1169 setattr(self.cls, k, v)
1171 self._apply_dataclasses_to_any_class(
1172 dataclass_setup_arguments, self.cls, annotations
1173 )
1175 @classmethod
1176 def _update_annotations_for_non_mapped_class(
1177 cls, klass: Type[_O]
1178 ) -> Mapping[str, _AnnotationScanType]:
1179 cls_annotations = util.get_annotations(klass)
1181 new_anno = {}
1182 for name, annotation in cls_annotations.items():
1183 if _is_mapped_annotation(annotation, klass, klass):
1184 extracted = _extract_mapped_subtype(
1185 annotation,
1186 klass,
1187 klass.__module__,
1188 name,
1189 type(None),
1190 required=False,
1191 is_dataclass_field=False,
1192 expect_mapped=False,
1193 )
1194 if extracted:
1195 inner, _ = extracted
1196 new_anno[name] = inner
1197 else:
1198 new_anno[name] = annotation
1199 return new_anno
1201 @classmethod
1202 def _apply_dataclasses_to_any_class(
1203 cls,
1204 dataclass_setup_arguments: _DataclassArguments,
1205 klass: Type[_O],
1206 use_annotations: Mapping[str, _AnnotationScanType],
1207 ) -> None:
1208 cls._assert_dc_arguments(dataclass_setup_arguments)
1210 dataclass_callable = dataclass_setup_arguments["dataclass_callable"]
1211 if dataclass_callable is _NoArg.NO_ARG:
1212 dataclass_callable = dataclasses.dataclass
1214 restored: Optional[Any]
1216 if use_annotations:
1217 # apply constructed annotations that should look "normal" to a
1218 # dataclasses callable, based on the fields present. This
1219 # means remove the Mapped[] container and ensure all Field
1220 # entries have an annotation
1221 restored = util.get_annotations(klass)
1222 klass.__annotations__ = cast("Dict[str, Any]", use_annotations)
1223 else:
1224 restored = None
1226 try:
1227 dataclass_callable( # type: ignore[call-overload]
1228 klass,
1229 **{ # type: ignore[call-overload,unused-ignore]
1230 k: v
1231 for k, v in dataclass_setup_arguments.items()
1232 if v is not _NoArg.NO_ARG and k != "dataclass_callable"
1233 },
1234 )
1235 except (TypeError, ValueError) as ex:
1236 raise exc.InvalidRequestError(
1237 f"Python dataclasses error encountered when creating "
1238 f"dataclass for {klass.__name__!r}: "
1239 f"{ex!r}. Please refer to Python dataclasses "
1240 "documentation for additional information.",
1241 code="dcte",
1242 ) from ex
1243 else:
1244 # as of Python 3.15, dataclasses no longer renders the
1245 # auto-generated class docstring immediately; it instead installs
1246 # a descriptor that renders the ``__init__`` signature the first
1247 # time ``__doc__`` is accessed (see
1248 # ``dataclasses._AutoDocstring``). As the ``finally:`` block
1249 # below puts the class' original annotations back, that deferred
1250 # render would no longer see the dataclass-oriented annotations
1251 # applied above, and would omit them from the docstring entirely.
1252 # Read the attribute now, while those annotations are still in
1253 # place, so the docstring we generate is the same on every Python
1254 # version.
1255 klass.__doc__
1257 finally:
1258 # restore original annotations outside of the dataclasses
1259 # process; for mixins and __abstract__ superclasses, SQLAlchemy
1260 # Declarative will need to see the Mapped[] container inside the
1261 # annotations in order to map subclasses
1262 if use_annotations:
1263 if restored is None:
1264 del klass.__annotations__
1265 else:
1266 klass.__annotations__ = restored # type: ignore[assignment] # noqa: E501
1268 @classmethod
1269 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None:
1270 allowed = {
1271 "init",
1272 "repr",
1273 "order",
1274 "eq",
1275 "unsafe_hash",
1276 "kw_only",
1277 "match_args",
1278 "dataclass_callable",
1279 }
1280 disallowed_args = set(arguments).difference(allowed)
1281 if disallowed_args:
1282 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args))
1283 raise exc.ArgumentError(
1284 f"Dataclass argument(s) {msg} are not accepted"
1285 )
1287 def _collect_annotation(
1288 self,
1289 name: str,
1290 raw_annotation: _AnnotationScanType,
1291 originating_class: Type[Any],
1292 expect_mapped: Optional[bool],
1293 attr_value: Any,
1294 ) -> Optional[_CollectedAnnotation]:
1295 if name in self.collected_annotations:
1296 return self.collected_annotations[name]
1298 if raw_annotation is None:
1299 return None
1301 is_dataclass = self.is_dataclass_prior_to_mapping
1302 allow_unmapped = self.allow_unmapped_annotations
1304 if expect_mapped is None:
1305 is_dataclass_field = isinstance(attr_value, dataclasses.Field)
1306 expect_mapped = (
1307 not is_dataclass_field
1308 and not allow_unmapped
1309 and (
1310 attr_value is None
1311 or isinstance(attr_value, _MappedAttribute)
1312 )
1313 )
1315 is_dataclass_field = False
1316 extracted = _extract_mapped_subtype(
1317 raw_annotation,
1318 self.cls,
1319 originating_class.__module__,
1320 name,
1321 type(attr_value),
1322 required=False,
1323 is_dataclass_field=is_dataclass_field,
1324 expect_mapped=expect_mapped and not is_dataclass,
1325 )
1326 if extracted is None:
1327 # ClassVar can come out here
1328 return None
1330 extracted_mapped_annotation, mapped_container = extracted
1332 if attr_value is None and not is_literal(extracted_mapped_annotation):
1333 for elem in get_args(extracted_mapped_annotation):
1334 if is_fwd_ref(
1335 elem, check_generic=True, check_for_plain_string=True
1336 ):
1337 elem = de_stringify_annotation(
1338 self.cls,
1339 elem,
1340 originating_class.__module__,
1341 include_generic=True,
1342 )
1343 # look in Annotated[...] for an ORM construct,
1344 # such as Annotated[int, mapped_column(primary_key=True)]
1345 if isinstance(elem, _IntrospectsAnnotations):
1346 attr_value = elem.found_in_pep593_annotated()
1348 self.collected_annotations[name] = ca = _CollectedAnnotation(
1349 raw_annotation,
1350 mapped_container,
1351 extracted_mapped_annotation,
1352 is_dataclass,
1353 attr_value,
1354 originating_class.__module__,
1355 originating_class,
1356 )
1357 return ca
1359 def _warn_for_decl_attributes(
1360 self, cls: Type[Any], key: str, c: Any
1361 ) -> None:
1362 if isinstance(c, expression.ColumnElement):
1363 util.warn(
1364 f"Attribute '{key}' on class {cls} appears to "
1365 "be a non-schema SQLAlchemy expression "
1366 "object; this won't be part of the declarative mapping. "
1367 "To map arbitrary expressions, use ``column_property()`` "
1368 "or a similar function such as ``deferred()``, "
1369 "``query_expression()`` etc. "
1370 )
1372 def _produce_column_copies(
1373 self,
1374 attributes_for_class: Callable[
1375 [], Iterable[Tuple[str, Any, Any, bool]]
1376 ],
1377 attribute_is_overridden: Callable[[str, Any], bool],
1378 fixed_table: bool,
1379 originating_class: Type[Any],
1380 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]:
1381 cls = self.cls
1382 dict_ = self.clsdict_view
1383 locally_collected_attributes = {}
1384 column_copies = self.column_copies
1385 # copy mixin columns to the mapped class
1387 for name, obj, annotation, is_dataclass in attributes_for_class():
1388 if (
1389 not fixed_table
1390 and obj is None
1391 and _is_mapped_annotation(annotation, cls, originating_class)
1392 ):
1393 # obj is None means this is the annotation only path
1395 if attribute_is_overridden(name, obj):
1396 # perform same "overridden" check as we do for
1397 # Column/MappedColumn, this is how a mixin col is not
1398 # applied to an inherited subclass that does not have
1399 # the mixin. the anno-only path added here for
1400 # #9564
1401 continue
1403 collected_annotation = self._collect_annotation(
1404 name, annotation, originating_class, True, obj
1405 )
1406 obj = (
1407 collected_annotation.attr_value
1408 if collected_annotation is not None
1409 else obj
1410 )
1411 if obj is None:
1412 obj = MappedColumn()
1414 locally_collected_attributes[name] = obj
1415 setattr(cls, name, obj)
1417 elif isinstance(obj, (Column, MappedColumn)):
1418 if attribute_is_overridden(name, obj):
1419 # if column has been overridden
1420 # (like by the InstrumentedAttribute of the
1421 # superclass), skip. don't collect the annotation
1422 # either (issue #8718)
1423 continue
1425 collected_annotation = self._collect_annotation(
1426 name, annotation, originating_class, True, obj
1427 )
1428 obj = (
1429 collected_annotation.attr_value
1430 if collected_annotation is not None
1431 else obj
1432 )
1434 if name not in dict_ and not (
1435 "__table__" in dict_
1436 and (getattr(obj, "name", None) or name)
1437 in dict_["__table__"].c
1438 ):
1439 if obj.foreign_keys:
1440 for fk in obj.foreign_keys:
1441 if (
1442 fk._table_column is not None
1443 and fk._table_column.table is None
1444 ):
1445 raise exc.InvalidRequestError(
1446 "Columns with foreign keys to "
1447 "non-table-bound "
1448 "columns must be declared as "
1449 "@declared_attr callables "
1450 "on declarative mixin classes. "
1451 "For dataclass "
1452 "field() objects, use a lambda:."
1453 )
1455 column_copies[obj] = copy_ = obj._copy()
1457 locally_collected_attributes[name] = copy_
1458 setattr(cls, name, copy_)
1460 return locally_collected_attributes
1462 def _extract_mappable_attributes(self) -> None:
1463 cls = self.cls
1464 collected_attributes = self.collected_attributes
1466 our_stuff = self.properties
1468 _include_dunders = self._include_dunders
1470 late_mapped = _get_immediate_cls_attr(
1471 cls, "_sa_decl_prepare_nocascade", strict=True
1472 )
1474 allow_unmapped_annotations = self.allow_unmapped_annotations
1475 expect_annotations_wo_mapped = (
1476 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping
1477 )
1479 look_for_dataclass_things = bool(self.dataclass_setup_arguments)
1481 for k in list(collected_attributes):
1482 if k in _include_dunders:
1483 continue
1485 value = collected_attributes[k]
1487 if _is_declarative_props(value):
1488 # @declared_attr in collected_attributes only occurs here for a
1489 # @declared_attr that's directly on the mapped class;
1490 # for a mixin, these have already been evaluated
1491 if value._cascading:
1492 util.warn(
1493 "Use of @declared_attr.cascading only applies to "
1494 "Declarative 'mixin' and 'abstract' classes. "
1495 "Currently, this flag is ignored on mapped class "
1496 "%s" % self.cls
1497 )
1499 value = getattr(cls, k)
1501 elif (
1502 isinstance(value, QueryableAttribute)
1503 and value.class_ is not cls
1504 and value.key != k
1505 ):
1506 # detect a QueryableAttribute that's already mapped being
1507 # assigned elsewhere in userland, turn into a synonym()
1508 value = SynonymProperty(value.key)
1509 setattr(cls, k, value)
1511 if (
1512 isinstance(value, tuple)
1513 and len(value) == 1
1514 and isinstance(value[0], (Column, _MappedAttribute))
1515 ):
1516 util.warn(
1517 "Ignoring declarative-like tuple value of attribute "
1518 "'%s': possibly a copy-and-paste error with a comma "
1519 "accidentally placed at the end of the line?" % k
1520 )
1521 continue
1522 elif look_for_dataclass_things and isinstance(
1523 value, dataclasses.Field
1524 ):
1525 # we collected a dataclass Field; dataclasses would have
1526 # set up the correct state on the class
1527 continue
1528 elif not isinstance(value, (Column, _DCAttributeOptions)):
1529 # using @declared_attr for some object that
1530 # isn't Column/MapperProperty/_DCAttributeOptions; remove
1531 # from the clsdict_view
1532 # and place the evaluated value onto the class.
1533 collected_attributes.pop(k)
1534 self._warn_for_decl_attributes(cls, k, value)
1535 if not late_mapped:
1536 setattr(cls, k, value)
1537 continue
1538 # we expect to see the name 'metadata' in some valid cases;
1539 # however at this point we see it's assigned to something trying
1540 # to be mapped, so raise for that.
1541 # TODO: should "registry" here be also? might be too late
1542 # to change that now (2.0 betas)
1543 elif k in ("metadata",):
1544 raise exc.InvalidRequestError(
1545 f"Attribute name '{k}' is reserved when using the "
1546 "Declarative API."
1547 )
1548 elif isinstance(value, Column):
1549 _undefer_column_name(
1550 k, self.column_copies.get(value, value) # type: ignore
1551 )
1552 else:
1553 if isinstance(value, _IntrospectsAnnotations):
1554 (
1555 annotation,
1556 mapped_container,
1557 extracted_mapped_annotation,
1558 is_dataclass,
1559 attr_value,
1560 originating_module,
1561 originating_class,
1562 ) = self.collected_annotations.get(
1563 k, (None, None, None, False, None, None, None)
1564 )
1566 # issue #8692 - don't do any annotation interpretation if
1567 # an annotation were present and a container such as
1568 # Mapped[] etc. were not used. If annotation is None,
1569 # do declarative_scan so that the property can raise
1570 # for required
1571 if (
1572 mapped_container is not None
1573 or annotation is None
1574 # issue #10516: need to do declarative_scan even with
1575 # a non-Mapped annotation if we are doing
1576 # __allow_unmapped__, for things like col.name
1577 # assignment
1578 or allow_unmapped_annotations
1579 ):
1580 try:
1581 value.declarative_scan(
1582 self,
1583 self.registry,
1584 cls,
1585 originating_module,
1586 k,
1587 mapped_container,
1588 annotation,
1589 extracted_mapped_annotation,
1590 is_dataclass,
1591 )
1592 except NameError as ne:
1593 raise orm_exc.MappedAnnotationError(
1594 f"Could not resolve all types within mapped "
1595 f'annotation: "{annotation}". Ensure all '
1596 f"types are written correctly and are "
1597 f"imported within the module in use."
1598 ) from ne
1599 else:
1600 # assert that we were expecting annotations
1601 # without Mapped[] were going to be passed.
1602 # otherwise an error should have been raised
1603 # by util._extract_mapped_subtype before we got here.
1604 assert expect_annotations_wo_mapped
1606 if isinstance(value, _DCAttributeOptions):
1607 if (
1608 value._has_dataclass_arguments
1609 and not look_for_dataclass_things
1610 ):
1611 if isinstance(value, MapperProperty):
1612 argnames = [
1613 "init",
1614 "default_factory",
1615 "repr",
1616 "default",
1617 "dataclass_metadata",
1618 ]
1619 else:
1620 argnames = [
1621 "init",
1622 "default_factory",
1623 "repr",
1624 "dataclass_metadata",
1625 ]
1627 args = {
1628 a
1629 for a in argnames
1630 if getattr(
1631 value._attribute_options, f"dataclasses_{a}"
1632 )
1633 is not _NoArg.NO_ARG
1634 }
1636 raise exc.ArgumentError(
1637 f"Attribute '{k}' on class {cls} includes "
1638 f"dataclasses argument(s): "
1639 f"{', '.join(sorted(repr(a) for a in args))} but "
1640 f"class does not specify "
1641 "SQLAlchemy native dataclass configuration."
1642 )
1644 if not isinstance(value, (MapperProperty, _MapsColumns)):
1645 # filter for _DCAttributeOptions objects that aren't
1646 # MapperProperty / mapped_column(). Currently this
1647 # includes AssociationProxy. pop it from the things
1648 # we're going to map and set it up as a descriptor
1649 # on the class.
1650 collected_attributes.pop(k)
1652 # Assoc Prox (or other descriptor object that may
1653 # use _DCAttributeOptions) is usually here, except if
1654 # 1. we're a
1655 # dataclass, dataclasses would have removed the
1656 # attr here or 2. assoc proxy is coming from a
1657 # superclass, we want it to be direct here so it
1658 # tracks state or 3. assoc prox comes from
1659 # declared_attr, uncommon case
1660 setattr(cls, k, value)
1661 continue
1663 our_stuff[k] = value
1665 def _extract_declared_columns(self) -> None:
1666 our_stuff = self.properties
1668 # extract columns from the class dict
1669 declared_columns = self.declared_columns
1670 column_ordering = self.column_ordering
1671 name_to_prop_key = collections.defaultdict(set)
1673 for key, c in list(our_stuff.items()):
1674 if isinstance(c, _MapsColumns):
1675 mp_to_assign = c.mapper_property_to_assign
1676 if mp_to_assign:
1677 our_stuff[key] = mp_to_assign
1678 else:
1679 # if no mapper property to assign, this currently means
1680 # this is a MappedColumn that will produce a Column for us
1681 del our_stuff[key]
1683 for col, sort_order in c.columns_to_assign:
1684 if not isinstance(c, CompositeProperty):
1685 name_to_prop_key[col.name].add(key)
1686 declared_columns.add(col)
1688 # we would assert this, however we want the below
1689 # warning to take effect instead. See #9630
1690 # assert col not in column_ordering
1692 column_ordering[col] = sort_order
1694 # if this is a MappedColumn and the attribute key we
1695 # have is not what the column has for its key, map the
1696 # Column explicitly under the attribute key name.
1697 # otherwise, Mapper will map it under the column key.
1698 if mp_to_assign is None and key != col.key:
1699 our_stuff[key] = col
1700 elif isinstance(c, Column):
1701 # undefer previously occurred here, and now occurs earlier.
1702 # ensure every column we get here has been named
1703 assert c.name is not None
1704 name_to_prop_key[c.name].add(key)
1705 declared_columns.add(c)
1706 # if the column is the same name as the key,
1707 # remove it from the explicit properties dict.
1708 # the normal rules for assigning column-based properties
1709 # will take over, including precedence of columns
1710 # in multi-column ColumnProperties.
1711 if key == c.key:
1712 del our_stuff[key]
1714 for name, keys in name_to_prop_key.items():
1715 if len(keys) > 1:
1716 util.warn(
1717 "On class %r, Column object %r named "
1718 "directly multiple times, "
1719 "only one will be used: %s. "
1720 "Consider using orm.synonym instead"
1721 % (self.classname, name, (", ".join(sorted(keys))))
1722 )
1724 def _setup_table(self, table: Optional[FromClause] = None) -> None:
1725 cls = self.cls
1726 cls_as_Decl = cast("MappedClassProtocol[Any]", cls)
1728 tablename = self.tablename
1729 table_args = self.table_args
1730 clsdict_view = self.clsdict_view
1731 declared_columns = self.declared_columns
1732 column_ordering = self.column_ordering
1734 manager = attributes.manager_of_class(cls)
1736 if (
1737 self.table_fn is None
1738 and "__table__" not in clsdict_view
1739 and table is None
1740 ):
1741 if hasattr(cls, "__table_cls__"):
1742 table_cls = cast(
1743 Type[Table],
1744 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore # noqa: E501
1745 )
1746 else:
1747 table_cls = Table
1749 if tablename is not None:
1750 args: Tuple[Any, ...] = ()
1751 table_kw: Dict[str, Any] = {}
1753 if table_args:
1754 if isinstance(table_args, dict):
1755 table_kw = table_args
1756 elif isinstance(table_args, tuple):
1757 if isinstance(table_args[-1], dict):
1758 args, table_kw = table_args[0:-1], table_args[-1]
1759 else:
1760 args = table_args
1762 autoload_with = clsdict_view.get("__autoload_with__")
1763 if autoload_with:
1764 table_kw["autoload_with"] = autoload_with
1766 autoload = clsdict_view.get("__autoload__")
1767 if autoload:
1768 table_kw["autoload"] = True
1770 sorted_columns = sorted(
1771 declared_columns,
1772 key=lambda c: column_ordering.get(c, 0),
1773 )
1774 table = self.set_cls_attribute(
1775 "__table__",
1776 table_cls(
1777 tablename,
1778 self._metadata_for_cls(manager),
1779 *sorted_columns,
1780 *args,
1781 **table_kw,
1782 ),
1783 )
1784 else:
1785 if table is None:
1786 if self.table_fn:
1787 table = self.set_cls_attribute(
1788 "__table__", self.table_fn()
1789 )
1790 else:
1791 table = cls_as_Decl.__table__
1792 if declared_columns:
1793 for c in declared_columns:
1794 if not table.c.contains_column(c):
1795 raise exc.ArgumentError(
1796 "Can't add additional column %r when "
1797 "specifying __table__" % c.key
1798 )
1800 self.local_table = table
1802 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData:
1803 meta: Optional[MetaData] = getattr(self.cls, "metadata", None)
1804 if meta is not None:
1805 return meta
1806 else:
1807 return manager.registry.metadata
1809 def _setup_inheriting_mapper(self, mapper_kw: _MapperKwArgs) -> None:
1810 cls = self.cls
1812 inherits = mapper_kw.get("inherits", None)
1814 if inherits is None:
1815 # since we search for classical mappings now, search for
1816 # multiple mapped bases as well and raise an error.
1817 inherits_search = []
1818 for base_ in cls.__bases__:
1819 c = _resolve_for_abstract_or_classical(base_)
1820 if c is None:
1821 continue
1823 if _is_supercls_for_inherits(c) and c not in inherits_search:
1824 inherits_search.append(c)
1826 if inherits_search:
1827 if len(inherits_search) > 1:
1828 raise exc.InvalidRequestError(
1829 "Class %s has multiple mapped bases: %r"
1830 % (cls, inherits_search)
1831 )
1832 inherits = inherits_search[0]
1833 elif isinstance(inherits, Mapper):
1834 inherits = inherits.class_
1836 self.inherits = inherits
1838 clsdict_view = self.clsdict_view
1839 if "__table__" not in clsdict_view and self.tablename is None:
1840 self.single = True
1842 def _setup_inheriting_columns(self, mapper_kw: _MapperKwArgs) -> None:
1843 table = self.local_table
1844 cls = self.cls
1845 table_args = self.table_args
1846 declared_columns = self.declared_columns
1848 if (
1849 table is None
1850 and self.inherits is None
1851 and not _get_immediate_cls_attr(cls, "__no_table__")
1852 ):
1853 raise exc.InvalidRequestError(
1854 "Class %r does not have a __table__ or __tablename__ "
1855 "specified and does not inherit from an existing "
1856 "table-mapped class." % cls
1857 )
1858 elif self.inherits:
1859 inherited_mapper_or_config = _declared_mapping_info(self.inherits)
1860 assert inherited_mapper_or_config is not None
1861 inherited_table = inherited_mapper_or_config.local_table
1862 inherited_persist_selectable = (
1863 inherited_mapper_or_config.persist_selectable
1864 )
1866 if table is None:
1867 # single table inheritance.
1868 # ensure no table args
1869 if table_args:
1870 raise exc.ArgumentError(
1871 "Can't place __table_args__ on an inherited class "
1872 "with no table."
1873 )
1875 # add any columns declared here to the inherited table.
1876 if declared_columns and not isinstance(inherited_table, Table):
1877 raise exc.ArgumentError(
1878 f"Can't declare columns on single-table-inherited "
1879 f"subclass {self.cls}; superclass {self.inherits} "
1880 "is not mapped to a Table"
1881 )
1883 for col in declared_columns:
1884 assert inherited_table is not None
1885 if col.name in inherited_table.c:
1886 if inherited_table.c[col.name] is col:
1887 continue
1888 raise exc.ArgumentError(
1889 f"Column '{col}' on class {cls.__name__} "
1890 f"conflicts with existing column "
1891 f"'{inherited_table.c[col.name]}'. If using "
1892 f"Declarative, consider using the "
1893 "use_existing_column parameter of mapped_column() "
1894 "to resolve conflicts."
1895 )
1896 if col.primary_key:
1897 raise exc.ArgumentError(
1898 "Can't place primary key columns on an inherited "
1899 "class with no table."
1900 )
1902 if TYPE_CHECKING:
1903 assert isinstance(inherited_table, Table)
1905 inherited_table.append_column(col)
1906 if (
1907 inherited_persist_selectable is not None
1908 and inherited_persist_selectable is not inherited_table
1909 ):
1910 inherited_persist_selectable._refresh_for_new_column(
1911 col
1912 )
1914 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None:
1915 properties = self.properties
1917 if self.mapper_args_fn:
1918 mapper_args = self.mapper_args_fn()
1919 else:
1920 mapper_args = {}
1922 if mapper_kw:
1923 mapper_args.update(mapper_kw)
1925 if "properties" in mapper_args:
1926 properties = dict(properties)
1927 properties.update(mapper_args["properties"])
1929 # make sure that column copies are used rather
1930 # than the original columns from any mixins
1931 for k in ("version_id_col", "polymorphic_on"):
1932 if k in mapper_args:
1933 v = mapper_args[k]
1934 mapper_args[k] = self.column_copies.get(v, v)
1936 if "primary_key" in mapper_args:
1937 mapper_args["primary_key"] = [
1938 self.column_copies.get(v, v)
1939 for v in util.to_list(mapper_args["primary_key"])
1940 ]
1942 if "inherits" in mapper_args:
1943 inherits_arg = mapper_args["inherits"]
1944 if isinstance(inherits_arg, Mapper):
1945 inherits_arg = inherits_arg.class_
1947 if inherits_arg is not self.inherits:
1948 raise exc.InvalidRequestError(
1949 "mapper inherits argument given for non-inheriting "
1950 "class %s" % (mapper_args["inherits"])
1951 )
1953 if self.inherits:
1954 mapper_args["inherits"] = self.inherits
1956 if self.inherits and not mapper_args.get("concrete", False):
1957 # note the superclass is expected to have a Mapper assigned and
1958 # not be a deferred config, as this is called within map()
1959 inherited_mapper = class_mapper(self.inherits, False)
1960 inherited_table = inherited_mapper.local_table
1962 # single or joined inheritance
1963 # exclude any cols on the inherited table which are
1964 # not mapped on the parent class, to avoid
1965 # mapping columns specific to sibling/nephew classes
1966 if "exclude_properties" not in mapper_args:
1967 mapper_args["exclude_properties"] = exclude_properties = {
1968 c.key
1969 for c in inherited_table.c
1970 if c not in inherited_mapper._columntoproperty
1971 }.union(inherited_mapper.exclude_properties or ())
1972 exclude_properties.difference_update(
1973 [c.key for c in self.declared_columns]
1974 )
1976 # look through columns in the current mapper that
1977 # are keyed to a propname different than the colname
1978 # (if names were the same, we'd have popped it out above,
1979 # in which case the mapper makes this combination).
1980 # See if the superclass has a similar column property.
1981 # If so, join them together.
1982 for k, col in list(properties.items()):
1983 if not isinstance(col, expression.ColumnElement):
1984 continue
1985 if k in inherited_mapper._props:
1986 p = inherited_mapper._props[k]
1987 if isinstance(p, ColumnProperty):
1988 # note here we place the subclass column
1989 # first. See [ticket:1892] for background.
1990 properties[k] = [col] + p.columns
1991 result_mapper_args = mapper_args.copy()
1992 result_mapper_args["properties"] = properties
1993 self.mapper_args = result_mapper_args
1995 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
1996 self._prepare_mapper_arguments(mapper_kw)
1997 if hasattr(self.cls, "__mapper_cls__"):
1998 mapper_cls = cast(
1999 "Type[Mapper[Any]]",
2000 util.unbound_method_to_callable(
2001 self.cls.__mapper_cls__ # type: ignore
2002 ),
2003 )
2004 else:
2005 mapper_cls = Mapper
2007 return self.set_cls_attribute(
2008 "__mapper__",
2009 mapper_cls(self.cls, self.local_table, **self.mapper_args),
2010 )
2013@util.preload_module("sqlalchemy.orm.decl_api")
2014def _as_dc_declaredattr(
2015 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str
2016) -> Any:
2017 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr.
2018 # we can't write it because field.metadata is immutable :( so we have
2019 # to go through extra trouble to compare these
2020 decl_api = util.preloaded.orm_decl_api
2021 obj = field_metadata[sa_dataclass_metadata_key]
2022 if callable(obj) and not isinstance(obj, decl_api.declared_attr):
2023 return decl_api.declared_attr(obj)
2024 else:
2025 return obj
2028class _DeferredMapperConfig(_ClassScanMapperConfig):
2029 _cls: weakref.ref[Type[Any]]
2031 is_deferred = True
2033 _configs: util.OrderedDict[
2034 weakref.ref[Type[Any]], _DeferredMapperConfig
2035 ] = util.OrderedDict()
2037 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
2038 pass
2040 @property
2041 def cls(self) -> Type[Any]:
2042 return self._cls() # type: ignore
2044 @cls.setter
2045 def cls(self, class_: Type[Any]) -> None:
2046 self._cls = weakref.ref(class_, self._remove_config_cls)
2047 self._configs[self._cls] = self
2049 @classmethod
2050 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None:
2051 cls._configs.pop(ref, None)
2053 @classmethod
2054 def has_cls(cls, class_: Type[Any]) -> bool:
2055 # 2.6 fails on weakref if class_ is an old style class
2056 return isinstance(class_, type) and weakref.ref(class_) in cls._configs
2058 @classmethod
2059 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn:
2060 if hasattr(class_, "_sa_raise_deferred_config"):
2061 class_._sa_raise_deferred_config()
2063 raise orm_exc.UnmappedClassError(
2064 class_,
2065 msg=(
2066 f"Class {orm_exc._safe_cls_name(class_)} has a deferred "
2067 "mapping on it. It is not yet usable as a mapped class."
2068 ),
2069 )
2071 @classmethod
2072 def config_for_cls(cls, class_: Type[Any]) -> _DeferredMapperConfig:
2073 return cls._configs[weakref.ref(class_)]
2075 @classmethod
2076 def classes_for_base(
2077 cls, base_cls: Type[Any], sort: bool = True
2078 ) -> List[_DeferredMapperConfig]:
2079 classes_for_base = [
2080 m
2081 for m, cls_ in [(m, m.cls) for m in cls._configs.values()]
2082 if cls_ is not None and issubclass(cls_, base_cls)
2083 ]
2085 if not sort:
2086 return classes_for_base
2088 all_m_by_cls = {m.cls: m for m in classes_for_base}
2090 tuples: List[Tuple[_DeferredMapperConfig, _DeferredMapperConfig]] = []
2091 for m_cls in all_m_by_cls:
2092 tuples.extend(
2093 (all_m_by_cls[base_cls], all_m_by_cls[m_cls])
2094 for base_cls in m_cls.__bases__
2095 if base_cls in all_m_by_cls
2096 )
2097 return list(topological.sort(tuples, classes_for_base))
2099 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2100 self._configs.pop(self._cls, None)
2101 return super().map(mapper_kw)
2104def _add_attribute(
2105 cls: Type[Any], key: str, value: MapperProperty[Any]
2106) -> None:
2107 """add an attribute to an existing declarative class.
2109 This runs through the logic to determine MapperProperty,
2110 adds it to the Mapper, adds a column to the mapped Table, etc.
2112 """
2114 if "__mapper__" in cls.__dict__:
2115 mapped_cls = cast("MappedClassProtocol[Any]", cls)
2117 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table:
2118 if isinstance(mc.__table__, Table):
2119 return mc.__table__
2120 raise exc.InvalidRequestError(
2121 f"Cannot add a new attribute to mapped class {mc.__name__!r} "
2122 "because it's not mapped against a table."
2123 )
2125 if isinstance(value, Column):
2126 _undefer_column_name(key, value)
2127 _table_or_raise(mapped_cls).append_column(
2128 value, replace_existing=True
2129 )
2130 mapped_cls.__mapper__.add_property(key, value)
2131 elif isinstance(value, _MapsColumns):
2132 mp = value.mapper_property_to_assign
2133 for col, _ in value.columns_to_assign:
2134 _undefer_column_name(key, col)
2135 _table_or_raise(mapped_cls).append_column(
2136 col, replace_existing=True
2137 )
2138 if not mp:
2139 mapped_cls.__mapper__.add_property(key, col)
2140 if mp:
2141 mapped_cls.__mapper__.add_property(key, mp)
2142 elif isinstance(value, MapperProperty):
2143 mapped_cls.__mapper__.add_property(key, value)
2144 elif isinstance(value, QueryableAttribute) and value.key != key:
2145 # detect a QueryableAttribute that's already mapped being
2146 # assigned elsewhere in userland, turn into a synonym()
2147 value = SynonymProperty(value.key)
2148 mapped_cls.__mapper__.add_property(key, value)
2149 else:
2150 type.__setattr__(cls, key, value)
2151 mapped_cls.__mapper__._expire_memoizations()
2152 else:
2153 type.__setattr__(cls, key, value)
2156def _del_attribute(cls: Type[Any], key: str) -> None:
2157 if (
2158 "__mapper__" in cls.__dict__
2159 and key in cls.__dict__
2160 and not cast(
2161 "MappedClassProtocol[Any]", cls
2162 ).__mapper__._dispose_called
2163 ):
2164 value = cls.__dict__[key]
2165 if isinstance(
2166 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute)
2167 ):
2168 raise NotImplementedError(
2169 "Can't un-map individual mapped attributes on a mapped class."
2170 )
2171 else:
2172 type.__delattr__(cls, key)
2173 cast(
2174 "MappedClassProtocol[Any]", cls
2175 ).__mapper__._expire_memoizations()
2176 else:
2177 type.__delattr__(cls, key)
2180def _declarative_constructor(self: Any, **kwargs: Any) -> None:
2181 """A simple constructor that allows initialization from kwargs.
2183 Sets attributes on the constructed instance using the names and
2184 values in ``kwargs``.
2186 Only keys that are present as
2187 attributes of the instance's class are allowed. These could be,
2188 for example, any mapped columns or relationships.
2189 """
2190 cls_ = type(self)
2191 for k in kwargs:
2192 if not hasattr(cls_, k):
2193 raise TypeError(
2194 "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
2195 )
2196 setattr(self, k, kwargs[k])
2199_declarative_constructor.__name__ = "__init__"
2202def _undefer_column_name(key: str, column: Column[Any]) -> None:
2203 if column.key is None:
2204 column.key = key
2205 if column.name is None:
2206 column.name = key