Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/decl_base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
8"""Internal implementation for declarative."""
10from __future__ import annotations
12import collections
13import dataclasses
14import itertools
15import re
16from typing import Any
17from typing import Callable
18from typing import cast
19from typing import Dict
20from typing import get_args
21from typing import Iterable
22from typing import List
23from typing import Mapping
24from typing import NamedTuple
25from typing import NoReturn
26from typing import Optional
27from typing import Protocol
28from typing import Sequence
29from typing import Tuple
30from typing import Type
31from typing import TYPE_CHECKING
32from typing import TypeVar
33from typing import Union
34import weakref
36from . import attributes
37from . import clsregistry
38from . import exc as orm_exc
39from . import instrumentation
40from . import mapperlib
41from ._typing import _O
42from ._typing import attr_is_internal_proxy
43from .attributes import InstrumentedAttribute
44from .attributes import QueryableAttribute
45from .base import _is_mapped_class
46from .base import InspectionAttr
47from .descriptor_props import CompositeProperty
48from .descriptor_props import SynonymProperty
49from .interfaces import _AttributeOptions
50from .interfaces import _DataclassArguments
51from .interfaces import _DCAttributeOptions
52from .interfaces import _IntrospectsAnnotations
53from .interfaces import _MappedAttribute
54from .interfaces import _MapsColumns
55from .interfaces import MapperProperty
56from .mapper import Mapper
57from .properties import ColumnProperty
58from .properties import MappedColumn
59from .util import _extract_mapped_subtype
60from .util import _is_mapped_annotation
61from .util import _metadata_for_cls as _metadata_for_cls_fn
62from .util import class_mapper
63from .util import de_stringify_annotation
64from .. import exc
65from .. import util
66from ..sql import expression
67from ..sql._annotated_cols import TypedColumns
68from ..sql.base import _NoArg
69from ..sql.schema import Column
70from ..sql.schema import MetaData
71from ..sql.schema import Table
72from ..util import topological
73from ..util.typing import _AnnotationScanType
74from ..util.typing import is_fwd_ref
75from ..util.typing import is_literal
77if TYPE_CHECKING:
78 from ._typing import _ClassDict
79 from ._typing import _RegistryType
80 from .base import Mapped
81 from .decl_api import declared_attr
82 from .instrumentation import ClassManager
83 from ..sql.elements import NamedColumn
84 from ..sql.selectable import FromClause
86_T = TypeVar("_T", bound=Any)
88_MapperKwArgs = Mapping[str, Any]
89_TableArgsType = Union[Tuple[Any, ...], Dict[str, Any]]
92class MappedClassProtocol(Protocol[_O]):
93 """A protocol representing a SQLAlchemy mapped class.
95 The protocol is generic on the type of class, use
96 ``MappedClassProtocol[Any]`` to allow any mapped class.
97 """
99 __name__: str
100 __mapper__: Mapper[_O]
101 __table__: FromClause
103 def __call__(self, **kw: Any) -> _O: ...
106class _DeclMappedClassProtocol(MappedClassProtocol[_O], Protocol):
107 "Internal more detailed version of ``MappedClassProtocol``."
109 metadata: MetaData
110 __tablename__: str
111 __mapper_args__: _MapperKwArgs
112 __table_args__: Optional[_TableArgsType]
114 _sa_apply_dc_transforms: Optional[_DataclassArguments]
116 def __declare_first__(self) -> None: ...
118 def __declare_last__(self) -> None: ...
121def _declared_mapping_info(
122 cls: Type[Any],
123) -> Optional[Union[_DeferredDeclarativeConfig, Mapper[Any]]]:
124 # deferred mapping
125 if _DeferredDeclarativeConfig.has_cls(cls):
126 return _DeferredDeclarativeConfig.config_for_cls(cls)
127 # regular mapping
128 elif _is_mapped_class(cls):
129 return class_mapper(cls, configure=False)
130 else:
131 return None
134def _is_supercls_for_inherits(cls: Type[Any]) -> bool:
135 """return True if this class will be used as a superclass to set in
136 'inherits'.
138 This includes deferred mapper configs that aren't mapped yet, however does
139 not include classes with _sa_decl_prepare_nocascade (e.g.
140 ``AbstractConcreteBase``); these concrete-only classes are not set up as
141 "inherits" until after mappers are configured using
142 mapper._set_concrete_base()
144 """
145 if _DeferredDeclarativeConfig.has_cls(cls):
146 return not _get_immediate_cls_attr(
147 cls, "_sa_decl_prepare_nocascade", strict=True
148 )
149 # regular mapping
150 elif _is_mapped_class(cls):
151 return True
152 else:
153 return False
156def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
157 if cls is object:
158 return None
160 sup: Optional[Type[Any]]
162 if cls.__dict__.get("__abstract__", False):
163 for base_ in cls.__bases__:
164 sup = _resolve_for_abstract_or_classical(base_)
165 if sup is not None:
166 return sup
167 else:
168 return None
169 else:
170 clsmanager = _dive_for_cls_manager(cls)
172 if clsmanager:
173 return clsmanager.class_
174 else:
175 return cls
178def _get_immediate_cls_attr(
179 cls: Type[Any], attrname: str, strict: bool = False
180) -> Optional[Any]:
181 """return an attribute of the class that is either present directly
182 on the class, e.g. not on a superclass, or is from a superclass but
183 this superclass is a non-mapped mixin, that is, not a descendant of
184 the declarative base and is also not classically mapped.
186 This is used to detect attributes that indicate something about
187 a mapped class independently from any mapped classes that it may
188 inherit from.
190 """
192 # the rules are different for this name than others,
193 # make sure we've moved it out. transitional
194 assert attrname != "__abstract__"
196 if not issubclass(cls, object):
197 return None
199 if attrname in cls.__dict__:
200 return getattr(cls, attrname)
202 for base in cls.__mro__[1:]:
203 _is_classical_inherits = _dive_for_cls_manager(base) is not None
205 if attrname in base.__dict__ and (
206 base is cls
207 or (
208 (base in cls.__bases__ if strict else True)
209 and not _is_classical_inherits
210 )
211 ):
212 return getattr(base, attrname)
213 else:
214 return None
217def _dive_for_cls_manager(cls: Type[_O]) -> Optional[ClassManager[_O]]:
218 # because the class manager registration is pluggable,
219 # we need to do the search for every class in the hierarchy,
220 # rather than just a simple "cls._sa_class_manager"
222 for base in cls.__mro__:
223 manager: Optional[ClassManager[_O]] = attributes.opt_manager_of_class(
224 base
225 )
226 if manager:
227 return manager
228 return None
231@util.preload_module("sqlalchemy.orm.decl_api")
232def _is_declarative_props(obj: Any) -> bool:
233 _declared_attr_common = util.preloaded.orm_decl_api._declared_attr_common
235 return isinstance(obj, (_declared_attr_common, util.classproperty))
238def _check_declared_props_nocascade(
239 obj: Any, name: str, cls: Type[_O]
240) -> bool:
241 if _is_declarative_props(obj):
242 if getattr(obj, "_cascading", False):
243 util.warn(
244 "@declared_attr.cascading is not supported on the %s "
245 "attribute on class %s. This attribute invokes for "
246 "subclasses in any case." % (name, cls)
247 )
248 return True
249 else:
250 return False
253class _ORMClassConfigurator:
254 """Object that configures a class that's potentially going to be
255 mapped, and/or turned into an ORM dataclass.
257 This is the base class for all the configurator objects.
259 """
261 __slots__ = ("cls", "classname", "__weakref__")
263 cls: Type[Any]
264 classname: str
266 def __init__(self, cls_: Type[Any]):
267 self.cls = util.assert_arg_type(cls_, type, "cls_")
268 self.classname = cls_.__name__
270 @classmethod
271 def _as_declarative(
272 cls, registry: _RegistryType, cls_: Type[Any], dict_: _ClassDict
273 ) -> Optional[_MapperConfig]:
274 manager = attributes.opt_manager_of_class(cls_)
275 if manager and manager.class_ is cls_:
276 raise exc.InvalidRequestError(
277 f"Class {cls_!r} already has been instrumented declaratively"
278 )
280 # allow subclassing an orm class with typed columns without
281 # generating an orm class
282 if cls_.__dict__.get("__abstract__", False) or issubclass(
283 cls_, TypedColumns
284 ):
285 return None
287 defer_map = _get_immediate_cls_attr(
288 cls_, "_sa_decl_prepare_nocascade", strict=True
289 ) or hasattr(cls_, "_sa_decl_prepare")
291 if defer_map:
292 return _DeferredDeclarativeConfig(registry, cls_, dict_)
293 else:
294 return _DeclarativeMapperConfig(registry, cls_, dict_)
296 @classmethod
297 def _as_unmapped_dataclass(
298 cls, cls_: Type[Any], dict_: _ClassDict
299 ) -> _UnmappedDataclassConfig:
300 return _UnmappedDataclassConfig(cls_, dict_)
302 @classmethod
303 def _mapper(
304 cls,
305 registry: _RegistryType,
306 cls_: Type[_O],
307 table: Optional[FromClause],
308 mapper_kw: _MapperKwArgs,
309 ) -> Mapper[_O]:
310 _ImperativeMapperConfig(registry, cls_, table, mapper_kw)
311 return cast("MappedClassProtocol[_O]", cls_).__mapper__
314class _MapperConfig(_ORMClassConfigurator):
315 """Configurator that configures a class that's potentially going to be
316 mapped, and optionally turned into a dataclass as well."""
318 __slots__ = (
319 "properties",
320 "declared_attr_reg",
321 )
323 properties: util.OrderedDict[
324 str,
325 Union[
326 Sequence[NamedColumn[Any]], NamedColumn[Any], MapperProperty[Any]
327 ],
328 ]
329 declared_attr_reg: Dict[declared_attr[Any], Any]
331 def __init__(
332 self,
333 registry: _RegistryType,
334 cls_: Type[Any],
335 ):
336 super().__init__(cls_)
337 self.properties = util.OrderedDict()
338 self.declared_attr_reg = {}
340 instrumentation.register_class(
341 self.cls,
342 finalize=False,
343 registry=registry,
344 declarative_scan=self,
345 init_method=registry.constructor,
346 )
348 def set_cls_attribute(self, attrname: str, value: _T) -> _T:
349 manager = instrumentation.manager_of_class(self.cls)
350 manager.install_member(attrname, value)
351 return value
353 def map(self, mapper_kw: _MapperKwArgs) -> Mapper[Any]:
354 raise NotImplementedError()
356 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
357 self.map(mapper_kw)
360class _ImperativeMapperConfig(_MapperConfig):
361 """Configurator that configures a class for an imperative mapping."""
363 __slots__ = ("local_table", "inherits")
365 def __init__(
366 self,
367 registry: _RegistryType,
368 cls_: Type[_O],
369 table: Optional[FromClause],
370 mapper_kw: _MapperKwArgs,
371 ):
372 super().__init__(registry, cls_)
374 self.local_table = self.set_cls_attribute("__table__", table)
376 with mapperlib._CONFIGURE_MUTEX:
377 clsregistry._add_class(
378 self.classname, self.cls, registry._class_registry
379 )
381 self._setup_inheritance(mapper_kw)
383 self._early_mapping(mapper_kw)
385 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
386 mapper_cls = Mapper
388 return self.set_cls_attribute(
389 "__mapper__",
390 mapper_cls(self.cls, self.local_table, **mapper_kw),
391 )
393 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
394 cls = self.cls
396 inherits = None
397 inherits_search = []
399 # since we search for classical mappings now, search for
400 # multiple mapped bases as well and raise an error.
401 for base_ in cls.__bases__:
402 c = _resolve_for_abstract_or_classical(base_)
403 if c is None:
404 continue
406 if _is_supercls_for_inherits(c) and c not in inherits_search:
407 inherits_search.append(c)
409 if inherits_search:
410 if len(inherits_search) > 1:
411 raise exc.InvalidRequestError(
412 "Class %s has multiple mapped bases: %r"
413 % (cls, inherits_search)
414 )
415 inherits = inherits_search[0]
417 self.inherits = inherits
420class _CollectedAnnotation(NamedTuple):
421 raw_annotation: _AnnotationScanType
422 mapped_container: Optional[Type[Mapped[Any]]]
423 extracted_mapped_annotation: Union[_AnnotationScanType, str]
424 is_dataclass: bool
425 attr_value: Any
426 originating_module: str
427 originating_class: Type[Any]
430class _ClassScanAbstractConfig(_ORMClassConfigurator):
431 """Abstract base for a configurator that configures a class for a
432 declarative mapping, or an unmapped ORM dataclass.
434 Defines scanning of pep-484 annotations as well as ORM dataclass
435 applicators
437 """
439 __slots__ = ()
441 clsdict_view: _ClassDict
442 collected_annotations: Dict[str, _CollectedAnnotation]
443 collected_attributes: Dict[str, Any]
445 is_dataclass_prior_to_mapping: bool
446 allow_unmapped_annotations: bool
448 dataclass_setup_arguments: Optional[_DataclassArguments]
449 """if the class has SQLAlchemy native dataclass parameters, where
450 we will turn the class into a dataclass within the declarative mapping
451 process.
453 """
455 allow_dataclass_fields: bool
456 """if true, look for dataclass-processed Field objects on the target
457 class as well as superclasses and extract ORM mapping directives from
458 the "metadata" attribute of each Field.
460 if False, dataclass fields can still be used, however they won't be
461 mapped.
463 """
465 _include_dunders = {
466 "__table__",
467 "__mapper_args__",
468 "__tablename__",
469 "__table_args__",
470 }
472 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)")
474 def _scan_attributes(self) -> None:
475 raise NotImplementedError()
477 def _setup_dataclasses_transforms(
478 self, *, enable_descriptor_defaults: bool, revert: bool = False
479 ) -> None:
480 dataclass_setup_arguments = self.dataclass_setup_arguments
481 if not dataclass_setup_arguments:
482 return
484 # can't use is_dataclass since it uses hasattr
485 if "__dataclass_fields__" in self.cls.__dict__:
486 raise exc.InvalidRequestError(
487 f"Class {self.cls} is already a dataclass; ensure that "
488 "base classes / decorator styles of establishing dataclasses "
489 "are not being mixed. "
490 "This can happen if a class that inherits from "
491 "'MappedAsDataclass', even indirectly, is been mapped with "
492 "'@registry.mapped_as_dataclass'"
493 )
495 # can't create a dataclass if __table__ is already there. This would
496 # fail an assertion when calling _get_arguments_for_make_dataclass:
497 # assert False, "Mapped[] received without a mapping declaration"
498 if "__table__" in self.cls.__dict__:
499 raise exc.InvalidRequestError(
500 f"Class {self.cls} already defines a '__table__'. "
501 "ORM Annotated Dataclasses do not support a pre-existing "
502 "'__table__' element"
503 )
505 raise_for_non_dc_attrs = collections.defaultdict(list)
507 def _allow_dataclass_field(
508 key: str, originating_class: Type[Any]
509 ) -> bool:
510 if (
511 originating_class is not self.cls
512 and "__dataclass_fields__" not in originating_class.__dict__
513 ):
514 raise_for_non_dc_attrs[originating_class].append(key)
516 return True
518 field_list = [
519 _AttributeOptions._get_arguments_for_make_dataclass(
520 self,
521 key,
522 anno,
523 mapped_container,
524 self.collected_attributes.get(key, _NoArg.NO_ARG),
525 dataclass_setup_arguments,
526 enable_descriptor_defaults,
527 )
528 for key, anno, mapped_container in (
529 (
530 key,
531 raw_anno,
532 mapped_container,
533 )
534 for key, (
535 raw_anno,
536 mapped_container,
537 mapped_anno,
538 is_dc,
539 attr_value,
540 originating_module,
541 originating_class,
542 ) in self.collected_annotations.items()
543 if _allow_dataclass_field(key, originating_class)
544 and (
545 key not in self.collected_attributes
546 # issue #9226; check for attributes that we've collected
547 # which are already instrumented, which we would assume
548 # mean we are in an ORM inheritance mapping and this
549 # attribute is already mapped on the superclass. Under
550 # no circumstance should any QueryableAttribute be sent to
551 # the dataclass() function; anything that's mapped should
552 # be Field and that's it
553 or not isinstance(
554 self.collected_attributes[key], QueryableAttribute
555 )
556 )
557 )
558 ]
559 if raise_for_non_dc_attrs:
560 for (
561 originating_class,
562 non_dc_attrs,
563 ) in raise_for_non_dc_attrs.items():
564 raise exc.InvalidRequestError(
565 f"When transforming {self.cls} to a dataclass, "
566 f"attribute(s) "
567 f"{', '.join(repr(key) for key in non_dc_attrs)} "
568 f"originates from superclass "
569 f"{originating_class}, which is not a dataclass. When "
570 f"declaring SQLAlchemy Declarative "
571 f"Dataclasses, ensure that all mixin classes and other "
572 f"superclasses which include attributes are also a "
573 f"subclass of MappedAsDataclass or make use of the "
574 f"@unmapped_dataclass decorator.",
575 code="dcmx",
576 )
578 if revert:
579 # the "revert" case is used only by an unmapped mixin class
580 # that is nonetheless using Mapped construct and needs to
581 # itself be a dataclass. revert_dict includes all non-dunder
582 # attributes, not just fields, in class __dict__ order, so that
583 # the finally: block below can restore that order
584 revert_dict = {
585 name: value
586 for name, value in self.cls.__dict__.items()
587 if not util.dunders_re.match(name)
588 }
589 else:
590 revert_dict = None
592 # get original annotations using ForwardRef for symbols that
593 # are unresolvable
594 orig_annotations = util.get_annotations(self.cls)
596 # build a new __annotations__ dict from the fields we have.
597 # this has to be done carefully since we have to maintain
598 # the correct order! wow
599 swap_annotations = {}
600 defaults = {}
602 for item in field_list:
603 if len(item) == 2:
604 name, tp = item
605 elif len(item) == 3:
606 name, tp, spec = item
607 defaults[name] = spec
608 else:
609 assert False
611 # add the annotation to the new dict we are creating.
612 # note that if name is in orig_annotations, we expect
613 # tp and orig_annotations[name] to be identical.
614 swap_annotations[name] = orig_annotations.get(name, tp)
616 for k, v in defaults.items():
617 setattr(self.cls, k, v)
619 self._assert_dc_arguments(dataclass_setup_arguments)
621 dataclass_callable = dataclass_setup_arguments["dataclass_callable"]
622 if dataclass_callable is _NoArg.NO_ARG:
623 dataclass_callable = dataclasses.dataclass
625 # create a merged __annotations__ dictionary, maintaining order
626 # as best we can:
628 # 1. merge all keys in orig_annotations that occur before
629 # we see any of our mapped fields (this can be attributes like
630 # __table_args__ etc.)
631 new_annotations = {
632 k: orig_annotations[k]
633 for k in itertools.takewhile(
634 lambda k: k not in swap_annotations, orig_annotations
635 )
636 }
638 # 2. then put in all the dataclass annotations we have
639 new_annotations |= swap_annotations
641 # 3. them merge all of orig_annotations which will add remaining
642 # keys
643 new_annotations |= orig_annotations
645 # 4. this becomes the new class annotations.
646 restore_anno = util.restore_annotations(self.cls, new_annotations)
648 try:
649 dataclass_callable( # type: ignore[call-overload]
650 self.cls,
651 **{ # type: ignore[call-overload,unused-ignore]
652 k: v
653 for k, v in dataclass_setup_arguments.items()
654 if v is not _NoArg.NO_ARG
655 and k not in ("dataclass_callable",)
656 },
657 )
658 except (TypeError, ValueError) as ex:
659 raise exc.InvalidRequestError(
660 f"Python dataclasses error encountered when creating "
661 f"dataclass for {self.cls.__name__!r}: "
662 f"{ex!r}. Please refer to Python dataclasses "
663 "documentation for additional information.",
664 code="dcte",
665 ) from ex
666 else:
667 # as of Python 3.15, dataclasses no longer renders the
668 # auto-generated class docstring immediately; it instead installs
669 # a descriptor that renders the ``__init__`` signature the first
670 # time ``__doc__`` is accessed (see
671 # ``dataclasses._AutoDocstring``). As the ``finally:`` block
672 # below puts the class' original annotations back, that deferred
673 # render would no longer see the dataclass-oriented annotations
674 # we swapped in above, and would omit them from the docstring
675 # entirely. Read the attribute now, while those annotations are
676 # still in place, so the docstring we generate is the same on
677 # every Python version.
678 self.cls.__doc__
680 finally:
681 if revert and revert_dict:
682 # used for mixin dataclasses; we have to restore the
683 # mapped_column(), relationship() etc. to the class so these
684 # take place for a mapped class scan. dataclasses has
685 # deleted the attribute for fields with no plain default and
686 # replaced it in place for the others, so delete and re-set
687 # every attribute, restoring declaration order for the scan,
688 # #13634
689 for k, v in revert_dict.items():
690 if k in self.cls.__dict__:
691 delattr(self.cls, k)
692 setattr(self.cls, k, v)
694 restore_anno()
696 def _collect_annotation(
697 self,
698 name: str,
699 raw_annotation: _AnnotationScanType,
700 originating_class: Type[Any],
701 expect_mapped: Optional[bool],
702 attr_value: Any,
703 ) -> Optional[_CollectedAnnotation]:
704 if name in self.collected_annotations:
705 return self.collected_annotations[name]
707 if raw_annotation is None:
708 return None
710 is_dataclass = self.is_dataclass_prior_to_mapping
711 allow_unmapped = self.allow_unmapped_annotations
713 if expect_mapped is None:
714 is_dataclass_field = isinstance(attr_value, dataclasses.Field)
715 expect_mapped = (
716 not is_dataclass_field
717 and not allow_unmapped
718 and (
719 attr_value is None
720 or isinstance(attr_value, _MappedAttribute)
721 )
722 )
724 is_dataclass_field = False
725 extracted = _extract_mapped_subtype(
726 raw_annotation,
727 self.cls,
728 originating_class.__module__,
729 name,
730 type(attr_value),
731 required=False,
732 is_dataclass_field=is_dataclass_field,
733 expect_mapped=expect_mapped and not is_dataclass,
734 )
735 if extracted is None:
736 # ClassVar can come out here
737 return None
739 extracted_mapped_annotation, mapped_container = extracted
741 if attr_value is None and not is_literal(extracted_mapped_annotation):
742 for elem in get_args(extracted_mapped_annotation):
743 if is_fwd_ref(
744 elem, check_generic=True, check_for_plain_string=True
745 ):
746 elem = de_stringify_annotation(
747 self.cls,
748 elem,
749 originating_class.__module__,
750 include_generic=True,
751 )
752 # look in Annotated[...] for an ORM construct,
753 # such as Annotated[int, mapped_column(primary_key=True)]
754 if isinstance(elem, _IntrospectsAnnotations):
755 attr_value = elem.found_in_pep593_annotated()
757 self.collected_annotations[name] = ca = _CollectedAnnotation(
758 raw_annotation,
759 mapped_container,
760 extracted_mapped_annotation,
761 is_dataclass,
762 attr_value,
763 originating_class.__module__,
764 originating_class,
765 )
766 return ca
768 @classmethod
769 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None:
770 allowed = {
771 "init",
772 "repr",
773 "order",
774 "eq",
775 "unsafe_hash",
776 "kw_only",
777 "match_args",
778 "dataclass_callable",
779 }
780 disallowed_args = set(arguments).difference(allowed)
781 if disallowed_args:
782 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args))
783 raise exc.ArgumentError(
784 f"Dataclass argument(s) {msg} are not accepted"
785 )
787 def _cls_attr_override_checker(
788 self, cls: Type[_O]
789 ) -> Callable[[str, Any], bool]:
790 """Produce a function that checks if a class has overridden an
791 attribute, taking SQLAlchemy-enabled dataclass fields into account.
793 """
795 if self.allow_dataclass_fields:
796 sa_dataclass_metadata_key = _get_immediate_cls_attr(
797 cls, "__sa_dataclass_metadata_key__"
798 )
799 else:
800 sa_dataclass_metadata_key = None
802 if not sa_dataclass_metadata_key:
804 def attribute_is_overridden(key: str, obj: Any) -> bool:
805 return getattr(cls, key, obj) is not obj
807 else:
808 all_datacls_fields = {
809 f.name: f.metadata[sa_dataclass_metadata_key]
810 for f in util.dataclass_fields(cls)
811 if sa_dataclass_metadata_key in f.metadata
812 }
813 local_datacls_fields = {
814 f.name: f.metadata[sa_dataclass_metadata_key]
815 for f in util.local_dataclass_fields(cls)
816 if sa_dataclass_metadata_key in f.metadata
817 }
819 absent = object()
821 def attribute_is_overridden(key: str, obj: Any) -> bool:
822 if _is_declarative_props(obj):
823 obj = obj.fget
825 # this function likely has some failure modes still if
826 # someone is doing a deep mixing of the same attribute
827 # name as plain Python attribute vs. dataclass field.
829 ret = local_datacls_fields.get(key, absent)
830 if _is_declarative_props(ret):
831 ret = ret.fget
833 if ret is obj:
834 return False
835 elif ret is not absent:
836 return True
838 all_field = all_datacls_fields.get(key, absent)
840 ret = getattr(cls, key, obj)
842 if ret is obj:
843 return False
845 # for dataclasses, this could be the
846 # 'default' of the field. so filter more specifically
847 # for an already-mapped InstrumentedAttribute
848 if ret is not absent and isinstance(
849 ret, InstrumentedAttribute
850 ):
851 return True
853 if all_field is obj:
854 return False
855 elif all_field is not absent:
856 return True
858 # can't find another attribute
859 return False
861 return attribute_is_overridden
863 def _cls_attr_resolver(
864 self, cls: Type[Any]
865 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]:
866 """produce a function to iterate the "attributes" of a class
867 which we want to consider for mapping, adjusting for SQLAlchemy fields
868 embedded in dataclass fields.
870 """
871 cls_annotations = util.get_annotations(cls)
873 cls_vars = vars(cls)
875 _include_dunders = self._include_dunders
876 _match_exclude_dunders = self._match_exclude_dunders
878 names = [
879 n
880 for n in util.merge_lists_w_ordering(
881 list(cls_vars), list(cls_annotations)
882 )
883 if not _match_exclude_dunders.match(n) or n in _include_dunders
884 ]
886 if self.allow_dataclass_fields:
887 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr(
888 cls, "__sa_dataclass_metadata_key__"
889 )
890 else:
891 sa_dataclass_metadata_key = None
893 if not sa_dataclass_metadata_key:
895 def local_attributes_for_class() -> (
896 Iterable[Tuple[str, Any, Any, bool]]
897 ):
898 return (
899 (
900 name,
901 cls_vars.get(name),
902 cls_annotations.get(name),
903 False,
904 )
905 for name in names
906 )
908 else:
909 dataclass_fields = {
910 field.name: field for field in util.local_dataclass_fields(cls)
911 }
913 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key
915 def local_attributes_for_class() -> (
916 Iterable[Tuple[str, Any, Any, bool]]
917 ):
918 for name in names:
919 field = dataclass_fields.get(name, None)
920 if field and sa_dataclass_metadata_key in field.metadata:
921 yield field.name, _as_dc_declaredattr(
922 field.metadata, fixed_sa_dataclass_metadata_key
923 ), cls_annotations.get(field.name), True
924 else:
925 yield name, cls_vars.get(name), cls_annotations.get(
926 name
927 ), False
929 return local_attributes_for_class
932class _DeclarativeMapperConfig(_MapperConfig, _ClassScanAbstractConfig):
933 """Configurator that will produce a declarative mapped class"""
935 __slots__ = (
936 "registry",
937 "local_table",
938 "persist_selectable",
939 "declared_columns",
940 "column_ordering",
941 "column_copies",
942 "table_args",
943 "tablename",
944 "mapper_args",
945 "mapper_args_fn",
946 "table_fn",
947 "inherits",
948 "single",
949 "clsdict_view",
950 "collected_attributes",
951 "collected_annotations",
952 "allow_dataclass_fields",
953 "dataclass_setup_arguments",
954 "is_dataclass_prior_to_mapping",
955 "allow_unmapped_annotations",
956 )
958 is_deferred = False
959 registry: _RegistryType
960 local_table: Optional[FromClause]
961 persist_selectable: Optional[FromClause]
962 declared_columns: util.OrderedSet[Column[Any]]
963 column_ordering: Dict[Column[Any], int]
964 column_copies: Dict[
965 Union[MappedColumn[Any], Column[Any]],
966 Union[MappedColumn[Any], Column[Any]],
967 ]
968 tablename: Optional[str]
969 mapper_args: Mapping[str, Any]
970 table_args: Optional[_TableArgsType]
971 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]]
972 inherits: Optional[Type[Any]]
973 single: bool
975 def __init__(
976 self,
977 registry: _RegistryType,
978 cls_: Type[_O],
979 dict_: _ClassDict,
980 ):
981 # grab class dict before the instrumentation manager has been added.
982 # reduces cycles
983 self.clsdict_view = (
984 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
985 )
986 super().__init__(registry, cls_)
987 self.registry = registry
988 self.persist_selectable = None
990 self.collected_attributes = {}
991 self.collected_annotations = {}
992 self.declared_columns = util.OrderedSet()
993 self.column_ordering = {}
994 self.column_copies = {}
995 self.single = False
996 self.dataclass_setup_arguments = dca = getattr(
997 self.cls, "_sa_apply_dc_transforms", None
998 )
1000 self.allow_unmapped_annotations = getattr(
1001 self.cls, "__allow_unmapped__", False
1002 ) or bool(self.dataclass_setup_arguments)
1004 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass(
1005 cls_
1006 )
1008 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__")
1010 # we don't want to consume Field objects from a not-already-dataclass.
1011 # the Field objects won't have their "name" or "type" populated,
1012 # and while it seems like we could just set these on Field as we
1013 # read them, Field is documented as "user read only" and we need to
1014 # stay far away from any off-label use of dataclasses APIs.
1015 if (not cld or dca) and sdk:
1016 raise exc.InvalidRequestError(
1017 "SQLAlchemy mapped dataclasses can't consume mapping "
1018 "information from dataclass.Field() objects if the immediate "
1019 "class is not already a dataclass."
1020 )
1022 # if already a dataclass, and __sa_dataclass_metadata_key__ present,
1023 # then also look inside of dataclass.Field() objects yielded by
1024 # dataclasses.get_fields(cls) when scanning for attributes
1025 self.allow_dataclass_fields = bool(sdk and cld)
1027 self._scan_attributes()
1029 self._setup_dataclasses_transforms(enable_descriptor_defaults=True)
1031 with mapperlib._CONFIGURE_MUTEX:
1032 clsregistry._add_class(
1033 self.classname, self.cls, registry._class_registry
1034 )
1036 self._setup_inheriting_mapper()
1038 self._extract_mappable_attributes()
1040 self._extract_declared_columns()
1042 self._setup_table()
1044 self._setup_inheriting_columns()
1046 self._early_mapping(util.EMPTY_DICT)
1048 def _scan_attributes(self) -> None:
1049 cls = self.cls
1051 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls)
1053 clsdict_view = self.clsdict_view
1054 collected_attributes = self.collected_attributes
1055 column_copies = self.column_copies
1056 _include_dunders = self._include_dunders
1057 mapper_args_fn = None
1058 table_args = inherited_table_args = None
1059 table_fn = None
1060 tablename = None
1061 fixed_table = "__table__" in clsdict_view
1063 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
1065 bases = []
1067 for base in cls.__mro__:
1068 # collect bases and make sure standalone columns are copied
1069 # to be the column they will ultimately be on the class,
1070 # so that declared_attr functions use the right columns.
1071 # need to do this all the way up the hierarchy first
1072 # (see #8190)
1074 class_mapped = base is not cls and _is_supercls_for_inherits(base)
1076 local_attributes_for_class = self._cls_attr_resolver(base)
1078 if not class_mapped and base is not cls:
1079 locally_collected_columns = self._produce_column_copies(
1080 local_attributes_for_class,
1081 attribute_is_overridden,
1082 fixed_table,
1083 base,
1084 )
1085 else:
1086 locally_collected_columns = {}
1088 bases.append(
1089 (
1090 base,
1091 class_mapped,
1092 local_attributes_for_class,
1093 locally_collected_columns,
1094 )
1095 )
1097 for (
1098 base,
1099 class_mapped,
1100 local_attributes_for_class,
1101 locally_collected_columns,
1102 ) in bases:
1103 # this transfer can also take place as we scan each name
1104 # for finer-grained control of how collected_attributes is
1105 # populated, as this is what impacts column ordering.
1106 # however it's simpler to get it out of the way here.
1107 collected_attributes.update(locally_collected_columns)
1109 for (
1110 name,
1111 obj,
1112 annotation,
1113 is_dataclass_field,
1114 ) in local_attributes_for_class():
1115 if name in _include_dunders:
1116 if name == "__mapper_args__":
1117 check_decl = _check_declared_props_nocascade(
1118 obj, name, cls
1119 )
1120 if not mapper_args_fn and (
1121 not class_mapped or check_decl
1122 ):
1123 # don't even invoke __mapper_args__ until
1124 # after we've determined everything about the
1125 # mapped table.
1126 # make a copy of it so a class-level dictionary
1127 # is not overwritten when we update column-based
1128 # arguments.
1129 def _mapper_args_fn() -> Dict[str, Any]:
1130 return dict(cls_as_Decl.__mapper_args__)
1132 mapper_args_fn = _mapper_args_fn
1134 elif name == "__tablename__":
1135 check_decl = _check_declared_props_nocascade(
1136 obj, name, cls
1137 )
1138 if not tablename and (not class_mapped or check_decl):
1139 tablename = cls_as_Decl.__tablename__
1140 elif name == "__table__":
1141 check_decl = _check_declared_props_nocascade(
1142 obj, name, cls
1143 )
1144 # if a @declared_attr using "__table__" is detected,
1145 # wrap up a callable to look for "__table__" from
1146 # the final concrete class when we set up a table.
1147 # this was fixed by
1148 # #11509, regression in 2.0 from version 1.4.
1149 if check_decl and not table_fn:
1150 # don't even invoke __table__ until we're ready
1151 def _table_fn() -> FromClause:
1152 return cls_as_Decl.__table__
1154 table_fn = _table_fn
1156 elif name == "__table_args__":
1157 check_decl = _check_declared_props_nocascade(
1158 obj, name, cls
1159 )
1160 if not table_args and (not class_mapped or check_decl):
1161 table_args = cls_as_Decl.__table_args__
1162 if not isinstance(
1163 table_args, (tuple, dict, type(None))
1164 ):
1165 raise exc.ArgumentError(
1166 "__table_args__ value must be a tuple, "
1167 "dict, or None"
1168 )
1169 if base is not cls:
1170 inherited_table_args = True
1171 else:
1172 # any other dunder names; should not be here
1173 # as we have tested for all four names in
1174 # _include_dunders
1175 assert False
1176 elif class_mapped:
1177 if _is_declarative_props(obj) and not obj._quiet:
1178 util.warn(
1179 "Regular (i.e. not __special__) "
1180 "attribute '%s.%s' uses @declared_attr, "
1181 "but owning class %s is mapped - "
1182 "not applying to subclass %s."
1183 % (base.__name__, name, base, cls)
1184 )
1186 continue
1187 elif base is not cls:
1188 # we're a mixin, abstract base, or something that is
1189 # acting like that for now.
1191 if isinstance(obj, (Column, MappedColumn)):
1192 # already copied columns to the mapped class.
1193 continue
1194 elif isinstance(obj, MapperProperty):
1195 raise exc.InvalidRequestError(
1196 "Mapper properties (i.e. deferred,"
1197 "column_property(), relationship(), etc.) must "
1198 "be declared as @declared_attr callables "
1199 "on declarative mixin classes. For dataclass "
1200 "field() objects, use a lambda:"
1201 )
1202 elif _is_declarative_props(obj):
1203 # tried to get overloads to tell this to
1204 # pylance, no luck
1205 assert obj is not None
1207 if obj._cascading:
1208 if name in clsdict_view:
1209 # unfortunately, while we can use the user-
1210 # defined attribute here to allow a clean
1211 # override, if there's another
1212 # subclass below then it still tries to use
1213 # this. not sure if there is enough
1214 # information here to add this as a feature
1215 # later on.
1216 util.warn(
1217 "Attribute '%s' on class %s cannot be "
1218 "processed due to "
1219 "@declared_attr.cascading; "
1220 "skipping" % (name, cls)
1221 )
1222 collected_attributes[name] = column_copies[obj] = (
1223 ret
1224 ) = obj.__get__(obj, cls)
1225 setattr(cls, name, ret)
1226 else:
1227 if is_dataclass_field:
1228 # access attribute using normal class access
1229 # first, to see if it's been mapped on a
1230 # superclass. note if the dataclasses.field()
1231 # has "default", this value can be anything.
1232 ret = getattr(cls, name, None)
1234 # so, if it's anything that's not ORM
1235 # mapped, assume we should invoke the
1236 # declared_attr
1237 if not isinstance(ret, InspectionAttr):
1238 ret = obj.fget()
1239 else:
1240 # access attribute using normal class access.
1241 # if the declared attr already took place
1242 # on a superclass that is mapped, then
1243 # this is no longer a declared_attr, it will
1244 # be the InstrumentedAttribute
1245 ret = getattr(cls, name)
1247 # correct for proxies created from hybrid_property
1248 # or similar. note there is no known case that
1249 # produces nested proxies, so we are only
1250 # looking one level deep right now.
1252 if (
1253 isinstance(ret, InspectionAttr)
1254 and attr_is_internal_proxy(ret)
1255 and not isinstance(
1256 ret.original_property, MapperProperty
1257 )
1258 ):
1259 ret = ret.descriptor
1261 collected_attributes[name] = column_copies[obj] = (
1262 ret
1263 )
1265 if (
1266 isinstance(ret, (Column, MapperProperty))
1267 and ret.doc is None
1268 ):
1269 ret.doc = obj.__doc__
1271 self._collect_annotation(
1272 name,
1273 obj._collect_return_annotation(),
1274 base,
1275 True,
1276 obj,
1277 )
1278 elif _is_mapped_annotation(annotation, cls, base):
1279 # Mapped annotation without any object.
1280 # product_column_copies should have handled this.
1281 # if future support for other MapperProperty,
1282 # then test if this name is already handled and
1283 # otherwise proceed to generate.
1284 if not fixed_table:
1285 assert (
1286 name in collected_attributes
1287 or attribute_is_overridden(name, None)
1288 )
1289 continue
1290 else:
1291 # here, the attribute is some other kind of
1292 # property that we assume is not part of the
1293 # declarative mapping. however, check for some
1294 # more common mistakes
1295 self._warn_for_decl_attributes(base, name, obj)
1296 elif is_dataclass_field and (
1297 name not in clsdict_view or clsdict_view[name] is not obj
1298 ):
1299 # here, we are definitely looking at the target class
1300 # and not a superclass. this is currently a
1301 # dataclass-only path. if the name is only
1302 # a dataclass field and isn't in local cls.__dict__,
1303 # put the object there.
1304 # assert that the dataclass-enabled resolver agrees
1305 # with what we are seeing
1307 assert not attribute_is_overridden(name, obj)
1309 if _is_declarative_props(obj):
1310 obj = obj.fget()
1312 collected_attributes[name] = obj
1313 self._collect_annotation(
1314 name, annotation, base, False, obj
1315 )
1316 else:
1317 collected_annotation = self._collect_annotation(
1318 name, annotation, base, None, obj
1319 )
1320 is_mapped = (
1321 collected_annotation is not None
1322 and collected_annotation.mapped_container is not None
1323 )
1324 generated_obj = (
1325 collected_annotation.attr_value
1326 if collected_annotation is not None
1327 else obj
1328 )
1329 if obj is None and not fixed_table and is_mapped:
1330 collected_attributes[name] = (
1331 generated_obj
1332 if generated_obj is not None
1333 else MappedColumn()
1334 )
1335 elif name in clsdict_view:
1336 collected_attributes[name] = obj
1337 # else if the name is not in the cls.__dict__,
1338 # don't collect it as an attribute.
1339 # we will see the annotation only, which is meaningful
1340 # both for mapping and dataclasses setup
1342 if inherited_table_args and not tablename:
1343 table_args = None
1345 self.table_args = table_args
1346 self.tablename = tablename
1347 self.mapper_args_fn = mapper_args_fn
1348 self.table_fn = table_fn
1350 @classmethod
1351 def _update_annotations_for_non_mapped_class(
1352 cls, klass: Type[_O]
1353 ) -> Mapping[str, _AnnotationScanType]:
1354 cls_annotations = util.get_annotations(klass)
1356 new_anno = {}
1357 for name, annotation in cls_annotations.items():
1358 if _is_mapped_annotation(annotation, klass, klass):
1359 extracted = _extract_mapped_subtype(
1360 annotation,
1361 klass,
1362 klass.__module__,
1363 name,
1364 type(None),
1365 required=False,
1366 is_dataclass_field=False,
1367 expect_mapped=False,
1368 )
1369 if extracted:
1370 inner, _ = extracted
1371 new_anno[name] = inner
1372 else:
1373 new_anno[name] = annotation
1374 return new_anno
1376 def _warn_for_decl_attributes(
1377 self, cls: Type[Any], key: str, c: Any
1378 ) -> None:
1379 if isinstance(c, expression.ColumnElement):
1380 util.warn(
1381 f"Attribute '{key}' on class {cls} appears to "
1382 "be a non-schema SQLAlchemy expression "
1383 "object; this won't be part of the declarative mapping. "
1384 "To map arbitrary expressions, use ``column_property()`` "
1385 "or a similar function such as ``deferred()``, "
1386 "``query_expression()`` etc. "
1387 )
1389 def _produce_column_copies(
1390 self,
1391 attributes_for_class: Callable[
1392 [], Iterable[Tuple[str, Any, Any, bool]]
1393 ],
1394 attribute_is_overridden: Callable[[str, Any], bool],
1395 fixed_table: bool,
1396 originating_class: Type[Any],
1397 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]:
1398 cls = self.cls
1399 dict_ = self.clsdict_view
1400 locally_collected_attributes = {}
1401 column_copies = self.column_copies
1402 # copy mixin columns to the mapped class
1404 for name, obj, annotation, is_dataclass in attributes_for_class():
1405 if (
1406 not fixed_table
1407 and obj is None
1408 and _is_mapped_annotation(annotation, cls, originating_class)
1409 ):
1410 # obj is None means this is the annotation only path
1412 if attribute_is_overridden(name, obj):
1413 # perform same "overridden" check as we do for
1414 # Column/MappedColumn, this is how a mixin col is not
1415 # applied to an inherited subclass that does not have
1416 # the mixin. the anno-only path added here for
1417 # #9564
1418 continue
1420 collected_annotation = self._collect_annotation(
1421 name, annotation, originating_class, True, obj
1422 )
1423 obj = (
1424 collected_annotation.attr_value
1425 if collected_annotation is not None
1426 else obj
1427 )
1428 if obj is None:
1429 obj = MappedColumn()
1431 locally_collected_attributes[name] = obj
1432 setattr(cls, name, obj)
1434 elif isinstance(obj, (Column, MappedColumn)):
1435 if attribute_is_overridden(name, obj):
1436 # if column has been overridden
1437 # (like by the InstrumentedAttribute of the
1438 # superclass), skip. don't collect the annotation
1439 # either (issue #8718)
1440 continue
1442 collected_annotation = self._collect_annotation(
1443 name, annotation, originating_class, True, obj
1444 )
1445 obj = (
1446 collected_annotation.attr_value
1447 if collected_annotation is not None
1448 else obj
1449 )
1451 if name not in dict_ and not (
1452 "__table__" in dict_
1453 and (getattr(obj, "name", None) or name)
1454 in dict_["__table__"].c
1455 ):
1456 if obj.foreign_keys:
1457 for fk in obj.foreign_keys:
1458 if (
1459 fk._table_column is not None
1460 and fk._table_column.table is None
1461 ):
1462 raise exc.InvalidRequestError(
1463 "Columns with foreign keys to "
1464 "non-table-bound "
1465 "columns must be declared as "
1466 "@declared_attr callables "
1467 "on declarative mixin classes. "
1468 "For dataclass "
1469 "field() objects, use a lambda:."
1470 )
1472 column_copies[obj] = copy_ = obj._copy()
1474 locally_collected_attributes[name] = copy_
1475 setattr(cls, name, copy_)
1477 return locally_collected_attributes
1479 def _extract_mappable_attributes(self) -> None:
1480 cls = self.cls
1481 collected_attributes = self.collected_attributes
1483 our_stuff = self.properties
1485 _include_dunders = self._include_dunders
1487 late_mapped = _get_immediate_cls_attr(
1488 cls, "_sa_decl_prepare_nocascade", strict=True
1489 )
1491 allow_unmapped_annotations = self.allow_unmapped_annotations
1492 expect_annotations_wo_mapped = (
1493 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping
1494 )
1496 look_for_dataclass_things = bool(self.dataclass_setup_arguments)
1498 for k in list(collected_attributes):
1499 if k in _include_dunders:
1500 continue
1502 value = collected_attributes[k]
1504 if _is_declarative_props(value):
1505 # @declared_attr in collected_attributes only occurs here for a
1506 # @declared_attr that's directly on the mapped class;
1507 # for a mixin, these have already been evaluated
1508 if value._cascading:
1509 util.warn(
1510 "Use of @declared_attr.cascading only applies to "
1511 "Declarative 'mixin' and 'abstract' classes. "
1512 "Currently, this flag is ignored on mapped class "
1513 "%s" % self.cls
1514 )
1516 value = getattr(cls, k)
1518 elif (
1519 isinstance(value, QueryableAttribute)
1520 and value.class_ is not cls
1521 and value.key != k
1522 ):
1523 # detect a QueryableAttribute that's already mapped being
1524 # assigned elsewhere in userland, turn into a synonym()
1525 value = SynonymProperty(value.key)
1526 setattr(cls, k, value)
1528 if k == "registry" or (
1529 k == "metadata" and not isinstance(value, MetaData)
1530 ):
1531 noun = {
1532 "metadata": "MetaData instance",
1533 "registry": "Declarative class registry",
1534 }
1535 util.warn(
1536 f"Attribute name '{k}' should be left reserved for the "
1537 f"{noun[k]} when using a Declarative class."
1538 )
1540 if (
1541 isinstance(value, tuple)
1542 and len(value) == 1
1543 and isinstance(value[0], (Column, _MappedAttribute))
1544 ):
1545 util.warn(
1546 "Ignoring declarative-like tuple value of attribute "
1547 "'%s': possibly a copy-and-paste error with a comma "
1548 "accidentally placed at the end of the line?" % k
1549 )
1550 continue
1551 elif look_for_dataclass_things and isinstance(
1552 value, dataclasses.Field
1553 ):
1554 # we collected a dataclass Field; dataclasses would have
1555 # set up the correct state on the class
1556 continue
1557 elif not isinstance(value, (Column, _DCAttributeOptions)):
1558 # using @declared_attr for some object that
1559 # isn't Column/MapperProperty/_DCAttributeOptions; remove
1560 # from the clsdict_view
1561 # and place the evaluated value onto the class.
1562 collected_attributes.pop(k)
1563 self._warn_for_decl_attributes(cls, k, value)
1564 if not late_mapped:
1565 setattr(cls, k, value)
1566 continue
1567 elif isinstance(value, Column):
1568 _undefer_column_name(
1569 k, self.column_copies.get(value, value) # type: ignore[arg-type] # noqa: E501
1570 )
1571 else:
1572 if isinstance(value, _IntrospectsAnnotations):
1573 (
1574 annotation,
1575 mapped_container,
1576 extracted_mapped_annotation,
1577 is_dataclass,
1578 attr_value,
1579 originating_module,
1580 originating_class,
1581 ) = self.collected_annotations.get(
1582 k, (None, None, None, False, None, None, None)
1583 )
1585 # issue #8692 - don't do any annotation interpretation if
1586 # an annotation were present and a container such as
1587 # Mapped[] etc. were not used. If annotation is None,
1588 # do declarative_scan so that the property can raise
1589 # for required
1590 if (
1591 mapped_container is not None
1592 or annotation is None
1593 # issue #10516: need to do declarative_scan even with
1594 # a non-Mapped annotation if we are doing
1595 # __allow_unmapped__, for things like col.name
1596 # assignment
1597 or allow_unmapped_annotations
1598 ):
1599 try:
1600 value.declarative_scan(
1601 self,
1602 self.registry,
1603 cls,
1604 originating_module,
1605 k,
1606 mapped_container,
1607 annotation,
1608 extracted_mapped_annotation,
1609 is_dataclass,
1610 )
1611 except NameError as ne:
1612 raise orm_exc.MappedAnnotationError(
1613 f"Could not resolve all types within mapped "
1614 f'annotation: "{annotation}". Ensure all '
1615 f"types are written correctly and are "
1616 f"imported within the module in use."
1617 ) from ne
1618 else:
1619 # assert that we were expecting annotations
1620 # without Mapped[] were going to be passed.
1621 # otherwise an error should have been raised
1622 # by util._extract_mapped_subtype before we got here.
1623 assert expect_annotations_wo_mapped
1625 if isinstance(value, _DCAttributeOptions):
1626 if (
1627 value._has_dataclass_arguments
1628 and not look_for_dataclass_things
1629 ):
1630 if isinstance(value, MapperProperty):
1631 argnames = [
1632 "init",
1633 "default_factory",
1634 "repr",
1635 "default",
1636 "dataclass_metadata",
1637 ]
1638 else:
1639 argnames = [
1640 "init",
1641 "default_factory",
1642 "repr",
1643 "dataclass_metadata",
1644 ]
1646 args = {
1647 a
1648 for a in argnames
1649 if getattr(
1650 value._attribute_options, f"dataclasses_{a}"
1651 )
1652 is not _NoArg.NO_ARG
1653 }
1655 raise exc.ArgumentError(
1656 f"Attribute '{k}' on class {cls} includes "
1657 f"dataclasses argument(s): "
1658 f"{', '.join(sorted(repr(a) for a in args))} but "
1659 f"class does not specify "
1660 "SQLAlchemy native dataclass configuration."
1661 )
1663 if not isinstance(value, (MapperProperty, _MapsColumns)):
1664 # filter for _DCAttributeOptions objects that aren't
1665 # MapperProperty / mapped_column(). Currently this
1666 # includes AssociationProxy. pop it from the things
1667 # we're going to map and set it up as a descriptor
1668 # on the class.
1669 collected_attributes.pop(k)
1671 # Assoc Prox (or other descriptor object that may
1672 # use _DCAttributeOptions) is usually here, except if
1673 # 1. we're a
1674 # dataclass, dataclasses would have removed the
1675 # attr here or 2. assoc proxy is coming from a
1676 # superclass, we want it to be direct here so it
1677 # tracks state or 3. assoc prox comes from
1678 # declared_attr, uncommon case
1679 setattr(cls, k, value)
1680 continue
1682 our_stuff[k] = value
1684 def _extract_declared_columns(self) -> None:
1685 our_stuff = self.properties
1687 # extract columns from the class dict
1688 declared_columns = self.declared_columns
1689 column_ordering = self.column_ordering
1690 name_to_prop_key = collections.defaultdict(set)
1692 for key, c in list(our_stuff.items()):
1693 if isinstance(c, _MapsColumns):
1694 mp_to_assign = c.mapper_property_to_assign
1695 if mp_to_assign:
1696 our_stuff[key] = mp_to_assign
1697 else:
1698 # if no mapper property to assign, this currently means
1699 # this is a MappedColumn that will produce a Column for us
1700 del our_stuff[key]
1702 for col, sort_order in c.columns_to_assign:
1703 if not isinstance(c, CompositeProperty):
1704 name_to_prop_key[col.name].add(key)
1705 declared_columns.add(col)
1707 # we would assert this, however we want the below
1708 # warning to take effect instead. See #9630
1709 # assert col not in column_ordering
1711 column_ordering[col] = sort_order
1713 # if this is a MappedColumn and the attribute key we
1714 # have is not what the column has for its key, map the
1715 # Column explicitly under the attribute key name.
1716 # otherwise, Mapper will map it under the column key.
1717 if mp_to_assign is None and key != col.key:
1718 our_stuff[key] = col
1719 elif isinstance(c, Column):
1720 # undefer previously occurred here, and now occurs earlier.
1721 # ensure every column we get here has been named
1722 assert c.name is not None
1723 name_to_prop_key[c.name].add(key)
1724 declared_columns.add(c)
1725 # if the column is the same name as the key,
1726 # remove it from the explicit properties dict.
1727 # the normal rules for assigning column-based properties
1728 # will take over, including precedence of columns
1729 # in multi-column ColumnProperties.
1730 if key == c.key:
1731 del our_stuff[key]
1733 for name, keys in name_to_prop_key.items():
1734 if len(keys) > 1:
1735 util.warn(
1736 "On class %r, Column object %r named "
1737 "directly multiple times, "
1738 "only one will be used: %s. "
1739 "Consider using orm.synonym instead"
1740 % (self.classname, name, (", ".join(sorted(keys))))
1741 )
1743 def _setup_table(self, table: Optional[FromClause] = None) -> None:
1744 cls = self.cls
1745 cls_as_Decl = cast("MappedClassProtocol[Any]", cls)
1747 tablename = self.tablename
1748 table_args = self.table_args
1749 clsdict_view = self.clsdict_view
1750 declared_columns = self.declared_columns
1751 column_ordering = self.column_ordering
1753 manager = attributes.manager_of_class(cls)
1755 if (
1756 self.table_fn is None
1757 and "__table__" not in clsdict_view
1758 and table is None
1759 ):
1760 if hasattr(cls, "__table_cls__"):
1761 table_cls = cast(
1762 Type[Table],
1763 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore[no-untyped-call] # noqa: E501
1764 )
1765 else:
1766 table_cls = Table
1768 if tablename is not None:
1769 args: Tuple[Any, ...] = ()
1770 table_kw: Dict[str, Any] = {}
1772 if table_args:
1773 if isinstance(table_args, dict):
1774 table_kw = table_args
1775 elif isinstance(table_args, tuple):
1776 if isinstance(table_args[-1], dict):
1777 args, table_kw = table_args[0:-1], table_args[-1]
1778 else:
1779 args = table_args
1781 autoload_with = clsdict_view.get("__autoload_with__")
1782 if autoload_with:
1783 table_kw["autoload_with"] = autoload_with
1785 autoload = clsdict_view.get("__autoload__")
1786 if autoload:
1787 table_kw["autoload"] = True
1789 sorted_columns = sorted(
1790 declared_columns,
1791 key=lambda c: column_ordering.get(c, 0),
1792 )
1793 table = self.set_cls_attribute(
1794 "__table__",
1795 table_cls(
1796 tablename,
1797 self._metadata_for_cls(manager),
1798 *sorted_columns,
1799 *args,
1800 **table_kw,
1801 ),
1802 )
1803 else:
1804 if table is None:
1805 if self.table_fn:
1806 table = self.set_cls_attribute(
1807 "__table__", self.table_fn()
1808 )
1809 else:
1810 table = cls_as_Decl.__table__
1811 if declared_columns:
1812 for c in declared_columns:
1813 if not table.c.contains_column(c):
1814 raise exc.ArgumentError(
1815 "Can't add additional column %r when "
1816 "specifying __table__" % c.key
1817 )
1819 self.local_table = table
1821 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData:
1822 return _metadata_for_cls_fn(self.cls, manager.registry)
1824 def _setup_inheriting_mapper(self) -> None:
1825 cls = self.cls
1827 inherits = None
1829 if inherits is None:
1830 # since we search for classical mappings now, search for
1831 # multiple mapped bases as well and raise an error.
1832 inherits_search = []
1833 for base_ in cls.__bases__:
1834 c = _resolve_for_abstract_or_classical(base_)
1835 if c is None:
1836 continue
1838 if _is_supercls_for_inherits(c) and c not in inherits_search:
1839 inherits_search.append(c)
1841 if inherits_search:
1842 if len(inherits_search) > 1:
1843 raise exc.InvalidRequestError(
1844 "Class %s has multiple mapped bases: %r"
1845 % (cls, inherits_search)
1846 )
1847 inherits = inherits_search[0]
1848 elif isinstance(inherits, Mapper):
1849 inherits = inherits.class_
1851 self.inherits = inherits
1853 clsdict_view = self.clsdict_view
1854 if "__table__" not in clsdict_view and self.tablename is None:
1855 self.single = True
1857 def _setup_inheriting_columns(self) -> None:
1858 table = self.local_table
1859 cls = self.cls
1860 table_args = self.table_args
1861 declared_columns = self.declared_columns
1863 if (
1864 table is None
1865 and self.inherits is None
1866 and not _get_immediate_cls_attr(cls, "__no_table__")
1867 ):
1868 raise exc.InvalidRequestError(
1869 "Class %r does not have a __table__ or __tablename__ "
1870 "specified and does not inherit from an existing "
1871 "table-mapped class." % cls
1872 )
1873 elif self.inherits:
1874 inherited_mapper_or_config = _declared_mapping_info(self.inherits)
1875 assert inherited_mapper_or_config is not None
1876 inherited_table = inherited_mapper_or_config.local_table
1877 inherited_persist_selectable = (
1878 inherited_mapper_or_config.persist_selectable
1879 )
1881 if table is None:
1882 # single table inheritance.
1883 # ensure no table args
1884 if table_args:
1885 raise exc.ArgumentError(
1886 "Can't place __table_args__ on an inherited class "
1887 "with no table."
1888 )
1890 # add any columns declared here to the inherited table.
1891 if declared_columns and not isinstance(inherited_table, Table):
1892 raise exc.ArgumentError(
1893 f"Can't declare columns on single-table-inherited "
1894 f"subclass {self.cls}; superclass {self.inherits} "
1895 "is not mapped to a Table"
1896 )
1898 for col in declared_columns:
1899 assert inherited_table is not None
1900 if col.name in inherited_table.c:
1901 if inherited_table.c[col.name] is col:
1902 continue
1903 raise exc.ArgumentError(
1904 f"Column '{col}' on class {cls.__name__} "
1905 f"conflicts with existing column "
1906 f"'{inherited_table.c[col.name]}'. If using "
1907 f"Declarative, consider using the "
1908 "use_existing_column parameter of mapped_column() "
1909 "to resolve conflicts."
1910 )
1911 if col.primary_key:
1912 raise exc.ArgumentError(
1913 "Can't place primary key columns on an inherited "
1914 "class with no table."
1915 )
1917 if TYPE_CHECKING:
1918 assert isinstance(inherited_table, Table)
1920 inherited_table.append_column(col)
1921 if (
1922 inherited_persist_selectable is not None
1923 and inherited_persist_selectable is not inherited_table
1924 ):
1925 inherited_persist_selectable._refresh_for_new_column(
1926 col
1927 )
1929 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None:
1930 properties = self.properties
1932 if self.mapper_args_fn:
1933 mapper_args = self.mapper_args_fn()
1934 else:
1935 mapper_args = {}
1937 if mapper_kw:
1938 mapper_args.update(mapper_kw)
1940 if "properties" in mapper_args:
1941 properties = dict(properties)
1942 properties.update(mapper_args["properties"])
1944 # make sure that column copies are used rather
1945 # than the original columns from any mixins
1946 for k in ("version_id_col", "polymorphic_on"):
1947 if k in mapper_args:
1948 v = mapper_args[k]
1949 mapper_args[k] = self.column_copies.get(v, v)
1951 if "primary_key" in mapper_args:
1952 mapper_args["primary_key"] = [
1953 self.column_copies.get(v, v)
1954 for v in util.to_list(mapper_args["primary_key"])
1955 ]
1957 if "inherits" in mapper_args:
1958 inherits_arg = mapper_args["inherits"]
1959 if isinstance(inherits_arg, Mapper):
1960 inherits_arg = inherits_arg.class_
1962 if inherits_arg is not self.inherits:
1963 raise exc.InvalidRequestError(
1964 "mapper inherits argument given for non-inheriting "
1965 "class %s" % (mapper_args["inherits"])
1966 )
1968 if self.inherits:
1969 mapper_args["inherits"] = self.inherits
1971 if self.inherits and not mapper_args.get("concrete", False):
1972 # note the superclass is expected to have a Mapper assigned and
1973 # not be a deferred config, as this is called within map()
1974 inherited_mapper = class_mapper(self.inherits, False)
1975 inherited_table = inherited_mapper.local_table
1977 # single or joined inheritance
1978 # exclude any cols on the inherited table which are
1979 # not mapped on the parent class, to avoid
1980 # mapping columns specific to sibling/nephew classes
1981 if "exclude_properties" not in mapper_args:
1982 mapper_args["exclude_properties"] = exclude_properties = {
1983 c.key
1984 for c in inherited_table.c
1985 if c not in inherited_mapper._columntoproperty
1986 }.union(inherited_mapper.exclude_properties or ())
1987 exclude_properties.difference_update(
1988 [c.key for c in self.declared_columns]
1989 )
1991 # look through columns in the current mapper that
1992 # are keyed to a propname different than the colname
1993 # (if names were the same, we'd have popped it out above,
1994 # in which case the mapper makes this combination).
1995 # See if the superclass has a similar column property.
1996 # If so, join them together.
1997 for k, col in list(properties.items()):
1998 if not isinstance(col, expression.ColumnElement):
1999 continue
2000 if k in inherited_mapper._props:
2001 p = inherited_mapper._props[k]
2002 if isinstance(p, ColumnProperty):
2003 # note here we place the subclass column
2004 # first. See [ticket:1892] for background.
2005 properties[k] = [col] + p.columns
2006 result_mapper_args = mapper_args.copy()
2007 result_mapper_args["properties"] = properties
2008 self.mapper_args = result_mapper_args
2010 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2011 self._prepare_mapper_arguments(mapper_kw)
2012 if hasattr(self.cls, "__mapper_cls__"):
2013 mapper_cls = cast(
2014 "Type[Mapper[Any]]",
2015 util.unbound_method_to_callable(
2016 self.cls.__mapper_cls__ # type: ignore[no-untyped-call]
2017 ),
2018 )
2019 else:
2020 mapper_cls = Mapper
2022 return self.set_cls_attribute(
2023 "__mapper__",
2024 mapper_cls(self.cls, self.local_table, **self.mapper_args),
2025 )
2028class _UnmappedDataclassConfig(_ClassScanAbstractConfig):
2029 """Configurator that will produce an unmapped dataclass."""
2031 __slots__ = (
2032 "clsdict_view",
2033 "collected_attributes",
2034 "collected_annotations",
2035 "allow_dataclass_fields",
2036 "dataclass_setup_arguments",
2037 "is_dataclass_prior_to_mapping",
2038 "allow_unmapped_annotations",
2039 )
2041 def __init__(
2042 self,
2043 cls_: Type[_O],
2044 dict_: _ClassDict,
2045 ):
2046 super().__init__(cls_)
2047 self.clsdict_view = (
2048 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT
2049 )
2050 self.dataclass_setup_arguments = getattr(
2051 self.cls, "_sa_apply_dc_transforms", None
2052 )
2054 self.is_dataclass_prior_to_mapping = dataclasses.is_dataclass(cls_)
2055 self.allow_dataclass_fields = False
2056 self.allow_unmapped_annotations = True
2057 self.collected_attributes = {}
2058 self.collected_annotations = {}
2060 self._scan_attributes()
2062 self._setup_dataclasses_transforms(
2063 enable_descriptor_defaults=False, revert=True
2064 )
2066 def _scan_attributes(self) -> None:
2067 cls = self.cls
2069 clsdict_view = self.clsdict_view
2070 collected_attributes = self.collected_attributes
2071 _include_dunders = self._include_dunders
2073 attribute_is_overridden = self._cls_attr_override_checker(self.cls)
2075 local_attributes_for_class = self._cls_attr_resolver(cls)
2076 for (
2077 name,
2078 obj,
2079 annotation,
2080 is_dataclass_field,
2081 ) in local_attributes_for_class():
2082 if name in _include_dunders:
2083 continue
2084 elif is_dataclass_field and (
2085 name not in clsdict_view or clsdict_view[name] is not obj
2086 ):
2087 # here, we are definitely looking at the target class
2088 # and not a superclass. this is currently a
2089 # dataclass-only path. if the name is only
2090 # a dataclass field and isn't in local cls.__dict__,
2091 # put the object there.
2092 # assert that the dataclass-enabled resolver agrees
2093 # with what we are seeing
2095 assert not attribute_is_overridden(name, obj)
2097 if _is_declarative_props(obj):
2098 obj = obj.fget()
2100 collected_attributes[name] = obj
2101 self._collect_annotation(name, annotation, cls, False, obj)
2102 else:
2103 self._collect_annotation(name, annotation, cls, None, obj)
2104 if name in clsdict_view:
2105 collected_attributes[name] = obj
2108@util.preload_module("sqlalchemy.orm.decl_api")
2109def _as_dc_declaredattr(
2110 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str
2111) -> Any:
2112 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr.
2113 # we can't write it because field.metadata is immutable :( so we have
2114 # to go through extra trouble to compare these
2115 decl_api = util.preloaded.orm_decl_api
2116 obj = field_metadata[sa_dataclass_metadata_key]
2117 if callable(obj) and not isinstance(obj, decl_api.declared_attr):
2118 return decl_api.declared_attr(obj)
2119 else:
2120 return obj
2123class _DeferredDeclarativeConfig(_DeclarativeMapperConfig):
2124 """Configurator that extends _DeclarativeMapperConfig to add a
2125 "deferred" step, to allow extensions like AbstractConcreteBase,
2126 DeferredMapping to partially set up a mapping that is "prepared"
2127 when table metadata is ready.
2129 """
2131 _cls: weakref.ref[Type[Any]]
2133 is_deferred = True
2135 _configs: util.OrderedDict[
2136 weakref.ref[Type[Any]], _DeferredDeclarativeConfig
2137 ] = util.OrderedDict()
2139 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None:
2140 pass
2142 @property
2143 def cls(self) -> Type[Any]:
2144 return self._cls() # type: ignore[return-value]
2146 @cls.setter
2147 def cls(self, class_: Type[Any]) -> None:
2148 self._cls = weakref.ref(class_, self._remove_config_cls)
2149 self._configs[self._cls] = self
2151 @classmethod
2152 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None:
2153 cls._configs.pop(ref, None)
2155 @classmethod
2156 def has_cls(cls, class_: Type[Any]) -> bool:
2157 # 2.6 fails on weakref if class_ is an old style class
2158 return isinstance(class_, type) and weakref.ref(class_) in cls._configs
2160 @classmethod
2161 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn:
2162 if hasattr(class_, "_sa_raise_deferred_config"):
2163 class_._sa_raise_deferred_config()
2165 raise orm_exc.UnmappedClassError(
2166 class_,
2167 msg=(
2168 f"Class {orm_exc._safe_cls_name(class_)} has a deferred "
2169 "mapping on it. It is not yet usable as a mapped class."
2170 ),
2171 )
2173 @classmethod
2174 def config_for_cls(cls, class_: Type[Any]) -> _DeferredDeclarativeConfig:
2175 return cls._configs[weakref.ref(class_)]
2177 @classmethod
2178 def classes_for_base(
2179 cls, base_cls: Type[Any], sort: bool = True
2180 ) -> List[_DeferredDeclarativeConfig]:
2181 classes_for_base = [
2182 m
2183 for m, cls_ in [(m, m.cls) for m in cls._configs.values()]
2184 if cls_ is not None and issubclass(cls_, base_cls)
2185 ]
2187 if not sort:
2188 return classes_for_base
2190 all_m_by_cls = {m.cls: m for m in classes_for_base}
2192 tuples: List[
2193 Tuple[_DeferredDeclarativeConfig, _DeferredDeclarativeConfig]
2194 ] = []
2195 for m_cls in all_m_by_cls:
2196 tuples.extend(
2197 (all_m_by_cls[base_cls], all_m_by_cls[m_cls])
2198 for base_cls in m_cls.__bases__
2199 if base_cls in all_m_by_cls
2200 )
2201 return list(topological.sort(tuples, classes_for_base))
2203 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]:
2204 self._configs.pop(self._cls, None)
2205 return super().map(mapper_kw)
2208def _add_attribute(
2209 cls: Type[Any], key: str, value: MapperProperty[Any]
2210) -> None:
2211 """add an attribute to an existing declarative class.
2213 This runs through the logic to determine MapperProperty,
2214 adds it to the Mapper, adds a column to the mapped Table, etc.
2216 """
2218 if "__mapper__" in cls.__dict__:
2219 mapped_cls = cast("MappedClassProtocol[Any]", cls)
2221 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table:
2222 if isinstance(mc.__table__, Table):
2223 return mc.__table__
2224 raise exc.InvalidRequestError(
2225 f"Cannot add a new attribute to mapped class {mc.__name__!r} "
2226 "because it's not mapped against a table."
2227 )
2229 if isinstance(value, Column):
2230 _undefer_column_name(key, value)
2231 _table_or_raise(mapped_cls).append_column(
2232 value, replace_existing=True
2233 )
2234 mapped_cls.__mapper__.add_property(key, value)
2235 elif isinstance(value, _MapsColumns):
2236 mp = value.mapper_property_to_assign
2237 for col, _ in value.columns_to_assign:
2238 _undefer_column_name(key, col)
2239 _table_or_raise(mapped_cls).append_column(
2240 col, replace_existing=True
2241 )
2242 if not mp:
2243 mapped_cls.__mapper__.add_property(key, col)
2244 if mp:
2245 mapped_cls.__mapper__.add_property(key, mp)
2246 elif isinstance(value, MapperProperty):
2247 mapped_cls.__mapper__.add_property(key, value)
2248 elif isinstance(value, QueryableAttribute) and value.key != key:
2249 # detect a QueryableAttribute that's already mapped being
2250 # assigned elsewhere in userland, turn into a synonym()
2251 value = SynonymProperty(value.key)
2252 mapped_cls.__mapper__.add_property(key, value)
2253 else:
2254 type.__setattr__(cls, key, value)
2255 mapped_cls.__mapper__._expire_memoizations()
2256 else:
2257 type.__setattr__(cls, key, value)
2260def _del_attribute(cls: Type[Any], key: str) -> None:
2261 if (
2262 "__mapper__" in cls.__dict__
2263 and key in cls.__dict__
2264 and not cast(
2265 "MappedClassProtocol[Any]", cls
2266 ).__mapper__._dispose_called
2267 ):
2268 value = cls.__dict__[key]
2269 if isinstance(
2270 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute)
2271 ):
2272 raise NotImplementedError(
2273 "Can't un-map individual mapped attributes on a mapped class."
2274 )
2275 else:
2276 type.__delattr__(cls, key)
2277 cast(
2278 "MappedClassProtocol[Any]", cls
2279 ).__mapper__._expire_memoizations()
2280 else:
2281 type.__delattr__(cls, key)
2284def _declarative_constructor(self: Any, **kwargs: Any) -> None:
2285 """A simple constructor that allows initialization from kwargs.
2287 Sets attributes on the constructed instance using the names and
2288 values in ``kwargs``.
2290 Only keys that are present as
2291 attributes of the instance's class are allowed. These could be,
2292 for example, any mapped columns or relationships.
2293 """
2294 cls_ = type(self)
2295 for k in kwargs:
2296 if not hasattr(cls_, k):
2297 raise TypeError(
2298 "%r is an invalid keyword argument for %s" % (k, cls_.__name__)
2299 )
2300 setattr(self, k, kwargs[k])
2303_declarative_constructor.__name__ = "__init__"
2306def _undefer_column_name(key: str, column: Column[Any]) -> None:
2307 if column.key is None:
2308 column.key = key
2309 if column.name is None:
2310 column.name = key