codekingpro/portable-devtools
114k
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
7
8"""Internal implementation for declarative."""
9
10from __future__ import annotations
11
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
32
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
73
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
83
84_T = TypeVar("_T", bound=Any)
85
86_MapperKwArgs = Mapping[str, Any]
87_TableArgsType = Union[Tuple[Any, ...], Dict[str, Any]]
88
89
90class MappedClassProtocol(Protocol[_O]):
91 """A protocol representing a SQLAlchemy mapped class.
92
93 The protocol is generic on the type of class, use
94 ``MappedClassProtocol[Any]`` to allow any mapped class.
95 """
96
97 __name__: str
98 __mapper__: Mapper[_O]
99 __table__: FromClause
100
101 def __call__(self, **kw: Any) -> _O: ...
102
103
104class _DeclMappedClassProtocol(MappedClassProtocol[_O], Protocol):
105 "Internal more detailed version of ``MappedClassProtocol``."
106
107 metadata: MetaData
108 __tablename__: str
109 __mapper_args__: _MapperKwArgs
110 __table_args__: Optional[_TableArgsType]
111
112 _sa_apply_dc_transforms: Optional[_DataclassArguments]
113
114 def __declare_first__(self) -> None: ...
115
116 def __declare_last__(self) -> None: ...
117
118
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]]]
128
129
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
141
142
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'.
146
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()
152
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
163
164
165def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
166 if cls is object:
167 return None
168
169 sup: Optional[Type[Any]]
170
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)
180
181 if clsmanager:
182 return clsmanager.class_
183 else:
184 return cls
185
186
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.
194
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.
198
199 """
200
201 # the rules are different for this name than others,
202 # make sure we've moved it out. transitional
203 assert attrname != "__abstract__"
204
205 if not issubclass(cls, object):
206 return None
207
208 if attrname in cls.__dict__:
209 return getattr(cls, attrname)
210
211 for base in cls.__mro__[1:]:
212 _is_classical_inherits = _dive_for_cls_manager(base) is not None
213
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
224
225
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"
230
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
238
239
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, {})
246
247
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__
256
257
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
261
262 return isinstance(obj, (_declared_attr_common, util.classproperty))
263
264
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
278
279
280class _MapperConfig:
281 __slots__ = (
282 "cls",
283 "classname",
284 "properties",
285 "declared_attr_reg",
286 "__weakref__",
287 )
288
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]
298
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 )
313
314 if cls_.__dict__.get("__abstract__", False):
315 return None
316
317 defer_map = _get_immediate_cls_attr(
318 cls_, "_sa_decl_prepare_nocascade", strict=True
319 ) or hasattr(cls_, "_sa_decl_prepare")
320
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 )
329
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 = {}
340
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 )
357
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
362
363 def map(self, mapper_kw: _MapperKwArgs = ...) -> Mapper[Any]:
364 raise NotImplementedError()
365
366 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
367 self.map(mapper_kw)
368
369
370class _ImperativeMapperConfig(_MapperConfig):
371 __slots__ = ("local_table", "inherits")
372
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)
381
382 self.local_table = self.set_cls_attribute("__table__", table)
383
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 )
389
390 self._setup_inheritance(mapper_kw)
391
392 self._early_mapping(mapper_kw)
393
394 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
395 mapper_cls = Mapper
396
397 return self.set_cls_attribute(
398 "__mapper__",
399 mapper_cls(self.cls, self.local_table, **mapper_kw),
400 )
401
402 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
403 cls = self.cls
404
405 inherits = mapper_kw.get("inherits", None)
406
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
415
416 if _is_supercls_for_inherits(c) and c not in inherits_search:
417 inherits_search.append(c)
418
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_
428
429 self.inherits = inherits
430
431
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]
440
441
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 )
465
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
485
486 is_dataclass_prior_to_mapping: bool
487 allow_unmapped_annotations: bool
488
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.
493
494 """
495
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.
500
501 if False, dataclass fields can still be used, however they won't be
502 mapped.
503
504 """
505
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
522
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 )
532
533 self.allow_unmapped_annotations = getattr(
534 self.cls, "__allow_unmapped__", False
535 ) or bool(self.dataclass_setup_arguments)
536
537 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
538 cls_
539 )
540
541 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
542
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 )
554
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)
559
560 self._setup_declared_events()
561
562 self._scan_attributes()
563
564 self._setup_dataclasses_transforms()
565
566 with mapperlib._CONFIGURE_MUTEX:
567 clsregistry.add_class(
568 self.classname, self.cls, registry._class_registry
569 )
570
571 self._setup_inheriting_mapper(mapper_kw)
572
573 self._extract_mappable_attributes()
574
575 self._extract_declared_columns()
576
577 self._setup_table(table)
578
579 self._setup_inheriting_columns(mapper_kw)
580
581 self._early_mapping(mapper_kw)
582
583 def _setup_declared_events(self) -> None:
584 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
585
586 @event.listens_for(Mapper, "after_configured")
587 def after_configured() -> None:
588 cast(
589 "_DeclMappedClassProtocol[Any]", self.cls
590 ).__declare_last__()
591
592 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
593
594 @event.listens_for(Mapper, "before_configured")
595 def before_configured() -> None:
596 cast(
597 "_DeclMappedClassProtocol[Any]", self.cls
598 ).__declare_first__()
599
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.
605
606 """
607
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
614
615 if not sa_dataclass_metadata_key:
616
617 def attribute_is_overridden(key: str, obj: Any) -> bool:
618 return getattr(cls, key, obj) is not obj
619
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 }
631
632 absent = object()
633
634 def attribute_is_overridden(key: str, obj: Any) -> bool:
635 if _is_declarative_props(obj):
636 obj = obj.fget
637
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.
641
642 ret = local_datacls_fields.get(key, absent)
643 if _is_declarative_props(ret):
644 ret = ret.fget
645
646 if ret is obj:
647 return False
648 elif ret is not absent:
649 return True
650
651 all_field = all_datacls_fields.get(key, absent)
652
653 ret = getattr(cls, key, obj)
654
655 if ret is obj:
656 return False
657
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
665
666 if all_field is obj:
667 return False
668 elif all_field is not absent:
669 return True
670
671 # can't find another attribute
672 return False
673
674 return attribute_is_overridden
675
676 _include_dunders = {
677 "__table__",
678 "__mapper_args__",
679 "__tablename__",
680 "__table_args__",
681 }
682
683 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
684
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.
691
692 """
693 cls_annotations = util.get_annotations(cls)
694
695 cls_vars = vars(cls)
696
697 _include_dunders = self._include_dunders
698 _match_exclude_dunders = self._match_exclude_dunders
699
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 ]
707
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
714
715 if not sa_dataclass_metadata_key:
716
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 )
729
730 else:
731 dataclass_fields = {
732 field.name: field for field in util.local_dataclass_fields(cls)
733 }
734
735 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
736
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
750
751 return local_attributes_for_class
752
753 def _scan_attributes(self) -> None:
754 cls = self.cls
755
756 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
757
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
767
768 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
769
770 bases = []
771
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)
778
779 class_mapped = base is not cls and _is_supercls_for_inherits(base)
780
781 local_attributes_for_class = self._cls_attr_resolver(base)
782
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 = {}
792
793 bases.append(
794 (
795 base,
796 class_mapped,
797 local_attributes_for_class,
798 locally_collected_columns,
799 )
800 )
801
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)
813
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__)
836
837 mapper_args_fn = _mapper_args_fn
838
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__
858
859 table_fn = _table_fn
860
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 )
890
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.
895
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
911
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)
938
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)
951
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.
956
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
965
966 collected_attributes[name] = column_copies[obj] = (
967 ret
968 )
969
970 if (
971 isinstance(ret, (Column, MapperProperty))
972 and ret.doc is None
973 ):
974 ret.doc = obj.__doc__
975
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
1011
1012 assert not attribute_is_overridden(name, obj)
1013
1014 if _is_declarative_props(obj):
1015 obj = obj.fget()
1016
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
1046
1047 if inherited_table_args and not tablename:
1048 table_args = None
1049
1050 self.table_args = table_args
1051 self.tablename = tablename
1052 self.mapper_args_fn = mapper_args_fn
1053 self.table_fn = table_fn
1054
1055 def _setup_dataclasses_transforms(self) -> None:
1056 dataclass_setup_arguments = self.dataclass_setup_arguments
1057 if not dataclass_setup_arguments:
1058 return
1059
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 )
1070
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 )
1080
1081 warn_for_non_dc_attrs = collections.defaultdict(list)
1082
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)
1091
1092 return True
1093
1094 manager = instrumentation.manager_of_class(self.cls)
1095 assert manager is not None
1096
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 ]
1135
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 )
1155
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
1167
1168 for k, v in defaults.items():
1169 setattr(self.cls, k, v)
1170
1171 self._apply_dataclasses_to_any_class(
1172 dataclass_setup_arguments, self.cls, annotations
1173 )
1174
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)
1180
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
1200