codekingpro/portable-devtools
114k
1# orm/decl_base.py
2# Copyright (C) 2005-2024 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 is_fwd_ref
69from ..util.typing import is_literal
70from ..util.typing import Protocol
71from ..util.typing import TypedDict
72from ..util.typing import typing_get_args
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 metadata: MetaData
107 __tablename__: str
108 __mapper_args__: _MapperKwArgs
109 __table_args__: Optional[_TableArgsType]
110
111 _sa_apply_dc_transforms: Optional[_DataclassArguments]
112
113 def __declare_first__(self) -> None: ...
114
115 def __declare_last__(self) -> None: ...
116
117
118class _DataclassArguments(TypedDict):
119 init: Union[_NoArg, bool]
120 repr: Union[_NoArg, bool]
121 eq: Union[_NoArg, bool]
122 order: Union[_NoArg, bool]
123 unsafe_hash: Union[_NoArg, bool]
124 match_args: Union[_NoArg, bool]
125 kw_only: Union[_NoArg, bool]
126 dataclass_callable: Union[_NoArg, Callable[..., Type[Any]]]
127
128
129def _declared_mapping_info(
130 cls: Type[Any],
131) -> Optional[Union[_DeferredMapperConfig, Mapper[Any]]]:
132 # deferred mapping
133 if _DeferredMapperConfig.has_cls(cls):
134 return _DeferredMapperConfig.config_for_cls(cls)
135 # regular mapping
136 elif _is_mapped_class(cls):
137 return class_mapper(cls, configure=False)
138 else:
139 return None
140
141
142def _is_supercls_for_inherits(cls: Type[Any]) -> bool:
143 """return True if this class will be used as a superclass to set in
144 'inherits'.
145
146 This includes deferred mapper configs that aren't mapped yet, however does
147 not include classes with _sa_decl_prepare_nocascade (e.g.
148 ``AbstractConcreteBase``); these concrete-only classes are not set up as
149 "inherits" until after mappers are configured using
150 mapper._set_concrete_base()
151
152 """
153 if _DeferredMapperConfig.has_cls(cls):
154 return not _get_immediate_cls_attr(
155 cls, "_sa_decl_prepare_nocascade", strict=True
156 )
157 # regular mapping
158 elif _is_mapped_class(cls):
159 return True
160 else:
161 return False
162
163
164def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
165 if cls is object:
166 return None
167
168 sup: Optional[Type[Any]]
169
170 if cls.__dict__.get("__abstract__", False):
171 for base_ in cls.__bases__:
172 sup = _resolve_for_abstract_or_classical(base_)
173 if sup is not None:
174 return sup
175 else:
176 return None
177 else:
178 clsmanager = _dive_for_cls_manager(cls)
179
180 if clsmanager:
181 return clsmanager.class_
182 else:
183 return cls
184
185
186def _get_immediate_cls_attr(
187 cls: Type[Any], attrname: str, strict: bool = False
188) -> Optional[Any]:
189 """return an attribute of the class that is either present directly
190 on the class, e.g. not on a superclass, or is from a superclass but
191 this superclass is a non-mapped mixin, that is, not a descendant of
192 the declarative base and is also not classically mapped.
193
194 This is used to detect attributes that indicate something about
195 a mapped class independently from any mapped classes that it may
196 inherit from.
197
198 """
199
200 # the rules are different for this name than others,
201 # make sure we've moved it out. transitional
202 assert attrname != "__abstract__"
203
204 if not issubclass(cls, object):
205 return None
206
207 if attrname in cls.__dict__:
208 return getattr(cls, attrname)
209
210 for base in cls.__mro__[1:]:
211 _is_classical_inherits = _dive_for_cls_manager(base) is not None
212
213 if attrname in base.__dict__ and (
214 base is cls
215 or (
216 (base in cls.__bases__ if strict else True)
217 and not _is_classical_inherits
218 )
219 ):
220 return getattr(base, attrname)
221 else:
222 return None
223
224
225def _dive_for_cls_manager(cls: Type[_O]) -> Optional[ClassManager[_O]]:
226 # because the class manager registration is pluggable,
227 # we need to do the search for every class in the hierarchy,
228 # rather than just a simple "cls._sa_class_manager"
229
230 for base in cls.__mro__:
231 manager: Optional[ClassManager[_O]] = attributes.opt_manager_of_class(
232 base
233 )
234 if manager:
235 return manager
236 return None
237
238
239def _as_declarative(
240 registry: _RegistryType, cls: Type[Any], dict_: _ClassDict
241) -> Optional[_MapperConfig]:
242 # declarative scans the class for attributes. no table or mapper
243 # args passed separately.
244 return _MapperConfig.setup_mapping(registry, cls, dict_, None, {})
245
246
247def _mapper(
248 registry: _RegistryType,
249 cls: Type[_O],
250 table: Optional[FromClause],
251 mapper_kw: _MapperKwArgs,
252) -> Mapper[_O]:
253 _ImperativeMapperConfig(registry, cls, table, mapper_kw)
254 return cast("MappedClassProtocol[_O]", cls).__mapper__
255
256
257@util.preload_module("sqlalchemy.orm.decl_api")
258def _is_declarative_props(obj: Any) -> bool:
259 _declared_attr_common = util.preloaded.orm_decl_api._declared_attr_common
260
261 return isinstance(obj, (_declared_attr_common, util.classproperty))
262
263
264def _check_declared_props_nocascade(
265 obj: Any, name: str, cls: Type[_O]
266) -> bool:
267 if _is_declarative_props(obj):
268 if getattr(obj, "_cascading", False):
269 util.warn(
270 "@declared_attr.cascading is not supported on the %s "
271 "attribute on class %s. This attribute invokes for "
272 "subclasses in any case." % (name, cls)
273 )
274 return True
275 else:
276 return False
277
278
279class _MapperConfig:
280 __slots__ = (
281 "cls",
282 "classname",
283 "properties",
284 "declared_attr_reg",
285 "__weakref__",
286 )
287
288 cls: Type[Any]
289 classname: str
290 properties: util.OrderedDict[
291 str,
292 Union[
293 Sequence[NamedColumn[Any]], NamedColumn[Any], MapperProperty[Any]
294 ],
295 ]
296 declared_attr_reg: Dict[declared_attr[Any], Any]
297
298 @classmethod
299 def setup_mapping(
300 cls,
301 registry: _RegistryType,
302 cls_: Type[_O],
303 dict_: _ClassDict,
304 table: Optional[FromClause],
305 mapper_kw: _MapperKwArgs,
306 ) -> Optional[_MapperConfig]:
307 manager = attributes.opt_manager_of_class(cls)
308 if manager and manager.class_ is cls_:
309 raise exc.InvalidRequestError(
310 f"Class {cls!r} already has been instrumented declaratively"
311 )
312
313 if cls_.__dict__.get("__abstract__", False):
314 return None
315
316 defer_map = _get_immediate_cls_attr(
317 cls_, "_sa_decl_prepare_nocascade", strict=True
318 ) or hasattr(cls_, "_sa_decl_prepare")
319
320 if defer_map:
321 return _DeferredMapperConfig(
322 registry, cls_, dict_, table, mapper_kw
323 )
324 else:
325 return _ClassScanMapperConfig(
326 registry, cls_, dict_, table, mapper_kw
327 )
328
329 def __init__(
330 self,
331 registry: _RegistryType,
332 cls_: Type[Any],
333 mapper_kw: _MapperKwArgs,
334 ):
335 self.cls = util.assert_arg_type(cls_, type, "cls_")
336 self.classname = cls_.__name__
337 self.properties = util.OrderedDict()
338 self.declared_attr_reg = {}
339
340 if not mapper_kw.get("non_primary", False):
341 instrumentation.register_class(
342 self.cls,
343 finalize=False,
344 registry=registry,
345 declarative_scan=self,
346 init_method=registry.constructor,
347 )
348 else:
349 manager = attributes.opt_manager_of_class(self.cls)
350 if not manager or not manager.is_mapped:
351 raise exc.InvalidRequestError(
352 "Class %s has no primary mapper configured. Configure "
353 "a primary mapper first before setting up a non primary "
354 "Mapper." % self.cls
355 )
356
357 def set_cls_attribute(self, attrname: str, value: _T) -> _T:
358 manager = instrumentation.manager_of_class(self.cls)
359 manager.install_member(attrname, value)
360 return value
361
362 def map(self, mapper_kw: _MapperKwArgs = ...) -> Mapper[Any]:
363 raise NotImplementedError()
364
365 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
366 self.map(mapper_kw)
367
368
369class _ImperativeMapperConfig(_MapperConfig):
370 __slots__ = ("local_table", "inherits")
371
372 def __init__(
373 self,
374 registry: _RegistryType,
375 cls_: Type[_O],
376 table: Optional[FromClause],
377 mapper_kw: _MapperKwArgs,
378 ):
379 super().__init__(registry, cls_, mapper_kw)
380
381 self.local_table = self.set_cls_attribute("__table__", table)
382
383 with mapperlib._CONFIGURE_MUTEX:
384 if not mapper_kw.get("non_primary", False):
385 clsregistry.add_class(
386 self.classname, self.cls, registry._class_registry
387 )
388
389 self._setup_inheritance(mapper_kw)
390
391 self._early_mapping(mapper_kw)
392
393 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
394 mapper_cls = Mapper
395
396 return self.set_cls_attribute(
397 "__mapper__",
398 mapper_cls(self.cls, self.local_table, **mapper_kw),
399 )
400
401 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
402 cls = self.cls
403
404 inherits = mapper_kw.get("inherits", None)
405
406 if inherits is None:
407 # since we search for classical mappings now, search for
408 # multiple mapped bases as well and raise an error.
409 inherits_search = []
410 for base_ in cls.__bases__:
411 c = _resolve_for_abstract_or_classical(base_)
412 if c is None:
413 continue
414
415 if _is_supercls_for_inherits(c) and c not in inherits_search:
416 inherits_search.append(c)
417
418 if inherits_search:
419 if len(inherits_search) > 1:
420 raise exc.InvalidRequestError(
421 "Class %s has multiple mapped bases: %r"
422 % (cls, inherits_search)
423 )
424 inherits = inherits_search[0]
425 elif isinstance(inherits, Mapper):
426 inherits = inherits.class_
427
428 self.inherits = inherits
429
430
431class _CollectedAnnotation(NamedTuple):
432 raw_annotation: _AnnotationScanType
433 mapped_container: Optional[Type[Mapped[Any]]]
434 extracted_mapped_annotation: Union[Type[Any], str]
435 is_dataclass: bool
436 attr_value: Any
437 originating_module: str
438 originating_class: Type[Any]
439
440
441class _ClassScanMapperConfig(_MapperConfig):
442 __slots__ = (
443 "registry",
444 "clsdict_view",
445 "collected_attributes",
446 "collected_annotations",
447 "local_table",
448 "persist_selectable",
449 "declared_columns",
450 "column_ordering",
451 "column_copies",
452 "table_args",
453 "tablename",
454 "mapper_args",
455 "mapper_args_fn",
456 "inherits",
457 "single",
458 "allow_dataclass_fields",
459 "dataclass_setup_arguments",
460 "is_dataclass_prior_to_mapping",
461 "allow_unmapped_annotations",
462 )
463
464 is_deferred = False
465 registry: _RegistryType
466 clsdict_view: _ClassDict
467 collected_annotations: Dict[str, _CollectedAnnotation]
468 collected_attributes: Dict[str, Any]
469 local_table: Optional[FromClause]
470 persist_selectable: Optional[FromClause]
471 declared_columns: util.OrderedSet[Column[Any]]
472 column_ordering: Dict[Column[Any], int]
473 column_copies: Dict[
474 Union[MappedColumn[Any], Column[Any]],
475 Union[MappedColumn[Any], Column[Any]],
476 ]
477 tablename: Optional[str]
478 mapper_args: Mapping[str, Any]
479 table_args: Optional[_TableArgsType]
480 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
481 inherits: Optional[Type[Any]]
482 single: bool
483
484 is_dataclass_prior_to_mapping: bool
485 allow_unmapped_annotations: bool
486
487 dataclass_setup_arguments: Optional[_DataclassArguments]
488 """if the class has SQLAlchemy native dataclass parameters, where
489 we will turn the class into a dataclass within the declarative mapping
490 process.
491
492 """
493
494 allow_dataclass_fields: bool
495 """if true, look for dataclass-processed Field objects on the target
496 class as well as superclasses and extract ORM mapping directives from
497 the "metadata" attribute of each Field.
498
499 if False, dataclass fields can still be used, however they won't be
500 mapped.
501
502 """
503
504 def __init__(
505 self,
506 registry: _RegistryType,
507 cls_: Type[_O],
508 dict_: _ClassDict,
509 table: Optional[FromClause],
510 mapper_kw: _MapperKwArgs,
511 ):
512 # grab class dict before the instrumentation manager has been added.
513 # reduces cycles
514 self.clsdict_view = (
515 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
516 )
517 super().__init__(registry, cls_, mapper_kw)
518 self.registry = registry
519 self.persist_selectable = None
520
521 self.collected_attributes = {}
522 self.collected_annotations = {}
523 self.declared_columns = util.OrderedSet()
524 self.column_ordering = {}
525 self.column_copies = {}
526 self.single = False
527 self.dataclass_setup_arguments = dca = getattr(
528 self.cls, "_sa_apply_dc_transforms", None
529 )
530
531 self.allow_unmapped_annotations = getattr(
532 self.cls, "__allow_unmapped__", False
533 ) or bool(self.dataclass_setup_arguments)
534
535 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
536 cls_
537 )
538
539 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
540
541 # we don't want to consume Field objects from a not-already-dataclass.
542 # the Field objects won't have their "name" or "type" populated,
543 # and while it seems like we could just set these on Field as we
544 # read them, Field is documented as "user read only" and we need to
545 # stay far away from any off-label use of dataclasses APIs.
546 if (not cld or dca) and sdk:
547 raise exc.InvalidRequestError(
548 "SQLAlchemy mapped dataclasses can't consume mapping "
549 "information from dataclass.Field() objects if the immediate "
550 "class is not already a dataclass."
551 )
552
553 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
554 # then also look inside of dataclass.Field() objects yielded by
555 # dataclasses.get_fields(cls) when scanning for attributes
556 self.allow_dataclass_fields = bool(sdk and cld)
557
558 self._setup_declared_events()
559
560 self._scan_attributes()
561
562 self._setup_dataclasses_transforms()
563
564 with mapperlib._CONFIGURE_MUTEX:
565 clsregistry.add_class(
566 self.classname, self.cls, registry._class_registry
567 )
568
569 self._setup_inheriting_mapper(mapper_kw)
570
571 self._extract_mappable_attributes()
572
573 self._extract_declared_columns()
574
575 self._setup_table(table)
576
577 self._setup_inheriting_columns(mapper_kw)
578
579 self._early_mapping(mapper_kw)
580
581 def _setup_declared_events(self) -> None:
582 if _get_immediate_cls_attr(self.cls, "__declare_last__"):
583
584 @event.listens_for(Mapper, "after_configured")
585 def after_configured() -> None:
586 cast(
587 "_DeclMappedClassProtocol[Any]", self.cls
588 ).__declare_last__()
589
590 if _get_immediate_cls_attr(self.cls, "__declare_first__"):
591
592 @event.listens_for(Mapper, "before_configured")
593 def before_configured() -> None:
594 cast(
595 "_DeclMappedClassProtocol[Any]", self.cls
596 ).__declare_first__()
597
598 def _cls_attr_override_checker(
599 self, cls: Type[_O]
600 ) -> Callable[[str, Any], bool]:
601 """Produce a function that checks if a class has overridden an
602 attribute, taking SQLAlchemy-enabled dataclass fields into account.
603
604 """
605
606 if self.allow_dataclass_fields:
607 sa_dataclass_metadata_key = _get_immediate_cls_attr(
608 cls, "__sa_dataclass_metadata_key__"
609 )
610 else:
611 sa_dataclass_metadata_key = None
612
613 if not sa_dataclass_metadata_key:
614
615 def attribute_is_overridden(key: str, obj: Any) -> bool:
616 return getattr(cls, key, obj) is not obj
617
618 else:
619 all_datacls_fields = {
620 f.name: f.metadata[sa_dataclass_metadata_key]
621 for f in util.dataclass_fields(cls)
622 if sa_dataclass_metadata_key in f.metadata
623 }
624 local_datacls_fields = {
625 f.name: f.metadata[sa_dataclass_metadata_key]
626 for f in util.local_dataclass_fields(cls)
627 if sa_dataclass_metadata_key in f.metadata
628 }
629
630 absent = object()
631
632 def attribute_is_overridden(key: str, obj: Any) -> bool:
633 if _is_declarative_props(obj):
634 obj = obj.fget
635
636 # this function likely has some failure modes still if
637 # someone is doing a deep mixing of the same attribute
638 # name as plain Python attribute vs. dataclass field.
639
640 ret = local_datacls_fields.get(key, absent)
641 if _is_declarative_props(ret):
642 ret = ret.fget
643
644 if ret is obj:
645 return False
646 elif ret is not absent:
647 return True
648
649 all_field = all_datacls_fields.get(key, absent)
650
651 ret = getattr(cls, key, obj)
652
653 if ret is obj:
654 return False
655
656 # for dataclasses, this could be the
657 # 'default' of the field. so filter more specifically
658 # for an already-mapped InstrumentedAttribute
659 if ret is not absent and isinstance(
660 ret, InstrumentedAttribute
661 ):
662 return True
663
664 if all_field is obj:
665 return False
666 elif all_field is not absent:
667 return True
668
669 # can't find another attribute
670 return False
671
672 return attribute_is_overridden
673
674 _include_dunders = {
675 "__table__",
676 "__mapper_args__",
677 "__tablename__",
678 "__table_args__",
679 }
680
681 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
682
683 def _cls_attr_resolver(
684 self, cls: Type[Any]
685 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
686 """produce a function to iterate the "attributes" of a class
687 which we want to consider for mapping, adjusting for SQLAlchemy fields
688 embedded in dataclass fields.
689
690 """
691 cls_annotations = util.get_annotations(cls)
692
693 cls_vars = vars(cls)
694
695 _include_dunders = self._include_dunders
696 _match_exclude_dunders = self._match_exclude_dunders
697
698 names = [
699 n
700 for n in util.merge_lists_w_ordering(
701 list(cls_vars), list(cls_annotations)
702 )
703 if not _match_exclude_dunders.match(n) or n in _include_dunders
704 ]
705
706 if self.allow_dataclass_fields:
707 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
708 cls, "__sa_dataclass_metadata_key__"
709 )
710 else:
711 sa_dataclass_metadata_key = None
712
713 if not sa_dataclass_metadata_key:
714
715 def local_attributes_for_class() -> (
716 Iterable[Tuple[str, Any, Any, bool]]
717 ):
718 return (
719 (
720 name,
721 cls_vars.get(name),
722 cls_annotations.get(name),
723 False,
724 )
725 for name in names
726 )
727
728 else:
729 dataclass_fields = {
730 field.name: field for field in util.local_dataclass_fields(cls)
731 }
732
733 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
734
735 def local_attributes_for_class() -> (
736 Iterable[Tuple[str, Any, Any, bool]]
737 ):
738 for name in names:
739 field = dataclass_fields.get(name, None)
740 if field and sa_dataclass_metadata_key in field.metadata:
741 yield field.name, _as_dc_declaredattr(
742 field.metadata, fixed_sa_dataclass_metadata_key
743 ), cls_annotations.get(field.name), True
744 else:
745 yield name, cls_vars.get(name), cls_annotations.get(
746 name
747 ), False
748
749 return local_attributes_for_class
750
751 def _scan_attributes(self) -> None:
752 cls = self.cls
753
754 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
755
756 clsdict_view = self.clsdict_view
757 collected_attributes = self.collected_attributes
758 column_copies = self.column_copies
759 _include_dunders = self._include_dunders
760 mapper_args_fn = None
761 table_args = inherited_table_args = None
762
763 tablename = None
764 fixed_table = "__table__" in clsdict_view
765
766 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
767
768 bases = []
769
770 for base in cls.__mro__:
771 # collect bases and make sure standalone columns are copied
772 # to be the column they will ultimately be on the class,
773 # so that declared_attr functions use the right columns.
774 # need to do this all the way up the hierarchy first
775 # (see #8190)
776
777 class_mapped = base is not cls and _is_supercls_for_inherits(base)
778
779 local_attributes_for_class = self._cls_attr_resolver(base)
780
781 if not class_mapped and base is not cls:
782 locally_collected_columns = self._produce_column_copies(
783 local_attributes_for_class,
784 attribute_is_overridden,
785 fixed_table,
786 base,
787 )
788 else:
789 locally_collected_columns = {}
790
791 bases.append(
792 (
793 base,
794 class_mapped,
795 local_attributes_for_class,
796 locally_collected_columns,
797 )
798 )
799
800 for (
801 base,
802 class_mapped,
803 local_attributes_for_class,
804 locally_collected_columns,
805 ) in bases:
806 # this transfer can also take place as we scan each name
807 # for finer-grained control of how collected_attributes is
808 # populated, as this is what impacts column ordering.
809 # however it's simpler to get it out of the way here.
810 collected_attributes.update(locally_collected_columns)
811
812 for (
813 name,
814 obj,
815 annotation,
816 is_dataclass_field,
817 ) in local_attributes_for_class():
818 if name in _include_dunders:
819 if name == "__mapper_args__":
820 check_decl = _check_declared_props_nocascade(
821 obj, name, cls
822 )
823 if not mapper_args_fn and (
824 not class_mapped or check_decl
825 ):
826 # don't even invoke __mapper_args__ until
827 # after we've determined everything about the
828 # mapped table.
829 # make a copy of it so a class-level dictionary
830 # is not overwritten when we update column-based
831 # arguments.
832 def _mapper_args_fn() -> Dict[str, Any]:
833 return dict(cls_as_Decl.__mapper_args__)
834
835 mapper_args_fn = _mapper_args_fn
836
837 elif name == "__tablename__":
838 check_decl = _check_declared_props_nocascade(
839 obj, name, cls
840 )
841 if not tablename and (not class_mapped or check_decl):
842 tablename = cls_as_Decl.__tablename__
843 elif name == "__table_args__":
844 check_decl = _check_declared_props_nocascade(
845 obj, name, cls
846 )
847 if not table_args and (not class_mapped or check_decl):
848 table_args = cls_as_Decl.__table_args__
849 if not isinstance(
850 table_args, (tuple, dict, type(None))
851 ):
852 raise exc.ArgumentError(
853 "__table_args__ value must be a tuple, "
854 "dict, or None"
855 )
856 if base is not cls:
857 inherited_table_args = True
858 else:
859 # skip all other dunder names, which at the moment
860 # should only be __table__
861 continue
862 elif class_mapped:
863 if _is_declarative_props(obj) and not obj._quiet:
864 util.warn(
865 "Regular (i.e. not __special__) "
866 "attribute '%s.%s' uses @declared_attr, "
867 "but owning class %s is mapped - "
868 "not applying to subclass %s."
869 % (base.__name__, name, base, cls)
870 )
871
872 continue
873 elif base is not cls:
874 # we're a mixin, abstract base, or something that is
875 # acting like that for now.
876
877 if isinstance(obj, (Column, MappedColumn)):
878 # already copied columns to the mapped class.
879 continue
880 elif isinstance(obj, MapperProperty):
881 raise exc.InvalidRequestError(
882 "Mapper properties (i.e. deferred,"
883 "column_property(), relationship(), etc.) must "
884 "be declared as @declared_attr callables "
885 "on declarative mixin classes. For dataclass "
886 "field() objects, use a lambda:"
887 )
888 elif _is_declarative_props(obj):
889 # tried to get overloads to tell this to
890 # pylance, no luck
891 assert obj is not None
892
893 if obj._cascading:
894 if name in clsdict_view:
895 # unfortunately, while we can use the user-
896 # defined attribute here to allow a clean
897 # override, if there's another
898 # subclass below then it still tries to use
899 # this. not sure if there is enough
900 # information here to add this as a feature
901 # later on.
902 util.warn(
903 "Attribute '%s' on class %s cannot be "
904 "processed due to "
905 "@declared_attr.cascading; "
906 "skipping" % (name, cls)
907 )
908 collected_attributes[name] = column_copies[obj] = (
909 ret
910 ) = obj.__get__(obj, cls)
911 setattr(cls, name, ret)
912 else:
913 if is_dataclass_field:
914 # access attribute using normal class access
915 # first, to see if it's been mapped on a
916 # superclass. note if the dataclasses.field()
917 # has "default", this value can be anything.
918 ret = getattr(cls, name, None)
919
920 # so, if it's anything that's not ORM
921 # mapped, assume we should invoke the
922 # declared_attr
923 if not isinstance(ret, InspectionAttr):
924 ret = obj.fget()
925 else:
926 # access attribute using normal class access.
927 # if the declared attr already took place
928 # on a superclass that is mapped, then
929 # this is no longer a declared_attr, it will
930 # be the InstrumentedAttribute
931 ret = getattr(cls, name)
932
933 # correct for proxies created from hybrid_property
934 # or similar. note there is no known case that
935 # produces nested proxies, so we are only
936 # looking one level deep right now.
937
938 if (
939 isinstance(ret, InspectionAttr)
940 and attr_is_internal_proxy(ret)
941 and not isinstance(
942 ret.original_property, MapperProperty
943 )
944 ):
945 ret = ret.descriptor
946
947 collected_attributes[name] = column_copies[obj] = (
948 ret
949 )
950
951 if (
952 isinstance(ret, (Column, MapperProperty))
953 and ret.doc is None
954 ):
955 ret.doc = obj.__doc__
956
957 self._collect_annotation(
958 name,
959 obj._collect_return_annotation(),
960 base,
961 True,
962 obj,
963 )
964 elif _is_mapped_annotation(annotation, cls, base):
965 # Mapped annotation without any object.
966 # product_column_copies should have handled this.
967 # if future support for other MapperProperty,
968 # then test if this name is already handled and
969 # otherwise proceed to generate.
970 if not fixed_table:
971 assert (
972 name in collected_attributes
973 or attribute_is_overridden(name, None)
974 )
975 continue
976 else:
977 # here, the attribute is some other kind of
978 # property that we assume is not part of the
979 # declarative mapping. however, check for some
980 # more common mistakes
981 self._warn_for_decl_attributes(base, name, obj)
982 elif is_dataclass_field and (
983 name not in clsdict_view or clsdict_view[name] is not obj
984 ):
985 # here, we are definitely looking at the target class
986 # and not a superclass. this is currently a
987 # dataclass-only path. if the name is only
988 # a dataclass field and isn't in local cls.__dict__,
989 # put the object there.
990 # assert that the dataclass-enabled resolver agrees
991 # with what we are seeing
992
993 assert not attribute_is_overridden(name, obj)
994
995 if _is_declarative_props(obj):
996 obj = obj.fget()
997
998 collected_attributes[name] = obj
999 self._collect_annotation(
1000 name, annotation, base, False, obj
1001 )
1002 else:
1003 collected_annotation = self._collect_annotation(
1004 name, annotation, base, None, obj
1005 )
1006 is_mapped = (
1007 collected_annotation is not None
1008 and collected_annotation.mapped_container is not None
1009 )
1010 generated_obj = (
1011 collected_annotation.attr_value
1012 if collected_annotation is not None
1013 else obj
1014 )
1015 if obj is None and not fixed_table and is_mapped:
1016 collected_attributes[name] = (
1017 generated_obj
1018 if generated_obj is not None
1019 else MappedColumn()
1020 )
1021 elif name in clsdict_view:
1022 collected_attributes[name] = obj
1023 # else if the name is not in the cls.__dict__,
1024 # don't collect it as an attribute.
1025 # we will see the annotation only, which is meaningful
1026 # both for mapping and dataclasses setup
1027
1028 if inherited_table_args and not tablename:
1029 table_args = None
1030
1031 self.table_args = table_args
1032 self.tablename = tablename
1033 self.mapper_args_fn = mapper_args_fn
1034
1035 def _setup_dataclasses_transforms(self) -> None:
1036 dataclass_setup_arguments = self.dataclass_setup_arguments
1037 if not dataclass_setup_arguments:
1038 return
1039
1040 # can't use is_dataclass since it uses hasattr
1041 if "__dataclass_fields__" in self.cls.__dict__:
1042 raise exc.InvalidRequestError(
1043 f"Class {self.cls} is already a dataclass; ensure that "
1044 "base classes / decorator styles of establishing dataclasses "
1045 "are not being mixed. "
1046 "This can happen if a class that inherits from "
1047 "'MappedAsDataclass', even indirectly, is been mapped with "
1048 "'@registry.mapped_as_dataclass'"
1049 )
1050
1051 warn_for_non_dc_attrs = collections.defaultdict(list)
1052
1053 def _allow_dataclass_field(
1054 key: str, originating_class: Type[Any]
1055 ) -> bool:
1056 if (
1057 originating_class is not self.cls
1058 and "__dataclass_fields__" not in originating_class.__dict__
1059 ):
1060 warn_for_non_dc_attrs[originating_class].append(key)
1061
1062 return True
1063
1064 manager = instrumentation.manager_of_class(self.cls)
1065 assert manager is not None
1066
1067 field_list = [
1068 _AttributeOptions._get_arguments_for_make_dataclass(
1069 key,
1070 anno,
1071 mapped_container,
1072 self.collected_attributes.get(key, _NoArg.NO_ARG),
1073 )
1074 for key, anno, mapped_container in (
1075 (
1076 key,
1077 mapped_anno if mapped_anno else raw_anno,
1078 mapped_container,
1079 )
1080 for key, (
1081 raw_anno,
1082 mapped_container,
1083 mapped_anno,
1084 is_dc,
1085 attr_value,
1086 originating_module,
1087 originating_class,
1088 ) in self.collected_annotations.items()
1089 if _allow_dataclass_field(key, originating_class)
1090 and (
1091 key not in self.collected_attributes
1092 # issue #9226; check for attributes that we've collected
1093 # which are already instrumented, which we would assume
1094 # mean we are in an ORM inheritance mapping and this
1095 # attribute is already mapped on the superclass. Under
1096 # no circumstance should any QueryableAttribute be sent to
1097 # the dataclass() function; anything that's mapped should
1098 # be Field and that's it
1099 or not isinstance(
1100 self.collected_attributes[key], QueryableAttribute
1101 )
1102 )
1103 )
1104 ]
1105
1106 if warn_for_non_dc_attrs:
1107 for (
1108 originating_class,
1109 non_dc_attrs,
1110 ) in warn_for_non_dc_attrs.items():
1111 util.warn_deprecated(
1112 f"When transforming {self.cls} to a dataclass, "
1113 f"attribute(s) "
1114 f"{', '.join(repr(key) for key in non_dc_attrs)} "
1115 f"originates from superclass "
1116 f"{originating_class}, which is not a dataclass. This "
1117 f"usage is deprecated and will raise an error in "
1118 f"SQLAlchemy 2.1. When declaring SQLAlchemy Declarative "
1119 f"Dataclasses, ensure that all mixin classes and other "
1120 f"superclasses which include attributes are also a "
1121 f"subclass of MappedAsDataclass.",
1122 "2.0",
1123 code="dcmx",
1124 )
1125
1126 annotations = {}
1127 defaults = {}
1128 for item in field_list:
1129 if len(item) == 2:
1130 name, tp = item
1131 elif len(item) == 3:
1132 name, tp, spec = item
1133 defaults[name] = spec
1134 else:
1135 assert False
1136 annotations[name] = tp
1137
1138 for k, v in defaults.items():
1139 setattr(self.cls, k, v)
1140
1141 self._apply_dataclasses_to_any_class(
1142 dataclass_setup_arguments, self.cls, annotations
1143 )
1144
1145 @classmethod
1146 def _update_annotations_for_non_mapped_class(
1147 cls, klass: Type[_O]
1148 ) -> Mapping[str, _AnnotationScanType]:
1149 cls_annotations = util.get_annotations(klass)
1150
1151 new_anno = {}
1152 for name, annotation in cls_annotations.items():
1153 if _is_mapped_annotation(annotation, klass, klass):
1154 extracted = _extract_mapped_subtype(
1155 annotation,
1156 klass,
1157 klass.__module__,
1158 name,
1159 type(None),
1160 required=False,
1161 is_dataclass_field=False,
1162 expect_mapped=False,
1163 )
1164 if extracted:
1165 inner, _ = extracted
1166 new_anno[name] = inner
1167 else:
1168 new_anno[name] = annotation
1169 return new_anno
1170
1171 @classmethod
1172 def _apply_dataclasses_to_any_class(
1173 cls,
1174 dataclass_setup_arguments: _DataclassArguments,
1175 klass: Type[_O],
1176 use_annotations: Mapping[str, _AnnotationScanType],
1177 ) -> None:
1178 cls._assert_dc_arguments(dataclass_setup_arguments)
1179
1180 dataclass_callable = dataclass_setup_arguments["dataclass_callable"]
1181 if dataclass_callable is _NoArg.NO_ARG:
1182 dataclass_callable = dataclasses.dataclass
1183
1184 restored: Optional[Any]
1185
1186 if use_annotations:
1187 # apply constructed annotations that should look "normal" to a
1188 # dataclasses callable, based on the fields present. This
1189 # means remove the Mapped[] container and ensure all Field
1190 # entries have an annotation
1191 restored = getattr(klass, "__annotations__", None)
1192 klass.__annotations__ = cast("Dict[str, Any]", use_annotations)
1193 else:
1194 restored = None
1195
1196 try:
1197 dataclass_callable(
1198 klass,
1199 **{
1200 k: v