Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/decl_base.py: 22%

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

915 statements  

1# orm/decl_base.py 

2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7 

8"""Internal implementation for declarative.""" 

9 

10from __future__ import annotations 

11 

12import collections 

13import dataclasses 

14import 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 

35 

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 event 

65from .. import exc 

66from .. import util 

67from ..sql import expression 

68from ..sql._annotated_cols import TypedColumns 

69from ..sql.base import _NoArg 

70from ..sql.schema import Column 

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 

76 

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.schema import MetaData 

85 from ..sql.selectable import FromClause 

86 

87_T = TypeVar("_T", bound=Any) 

88 

89_MapperKwArgs = Mapping[str, Any] 

90_TableArgsType = Union[Tuple[Any, ...], Dict[str, Any]] 

91 

92 

93class MappedClassProtocol(Protocol[_O]): 

94 """A protocol representing a SQLAlchemy mapped class. 

95 

96 The protocol is generic on the type of class, use 

97 ``MappedClassProtocol[Any]`` to allow any mapped class. 

98 """ 

99 

100 __name__: str 

101 __mapper__: Mapper[_O] 

102 __table__: FromClause 

103 

104 def __call__(self, **kw: Any) -> _O: ... 

105 

106 

107class _DeclMappedClassProtocol(MappedClassProtocol[_O], Protocol): 

108 "Internal more detailed version of ``MappedClassProtocol``." 

109 

110 metadata: MetaData 

111 __tablename__: str 

112 __mapper_args__: _MapperKwArgs 

113 __table_args__: Optional[_TableArgsType] 

114 

115 _sa_apply_dc_transforms: Optional[_DataclassArguments] 

116 

117 def __declare_first__(self) -> None: ... 

118 

119 def __declare_last__(self) -> None: ... 

120 

121 

122def _declared_mapping_info( 

123 cls: Type[Any], 

124) -> Optional[Union[_DeferredDeclarativeConfig, Mapper[Any]]]: 

125 # deferred mapping 

126 if _DeferredDeclarativeConfig.has_cls(cls): 

127 return _DeferredDeclarativeConfig.config_for_cls(cls) 

128 # regular mapping 

129 elif _is_mapped_class(cls): 

130 return class_mapper(cls, configure=False) 

131 else: 

132 return None 

133 

134 

135def _is_supercls_for_inherits(cls: Type[Any]) -> bool: 

136 """return True if this class will be used as a superclass to set in 

137 'inherits'. 

138 

139 This includes deferred mapper configs that aren't mapped yet, however does 

140 not include classes with _sa_decl_prepare_nocascade (e.g. 

141 ``AbstractConcreteBase``); these concrete-only classes are not set up as 

142 "inherits" until after mappers are configured using 

143 mapper._set_concrete_base() 

144 

145 """ 

146 if _DeferredDeclarativeConfig.has_cls(cls): 

147 return not _get_immediate_cls_attr( 

148 cls, "_sa_decl_prepare_nocascade", strict=True 

149 ) 

150 # regular mapping 

151 elif _is_mapped_class(cls): 

152 return True 

153 else: 

154 return False 

155 

156 

157def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]: 

158 if cls is object: 

159 return None 

160 

161 sup: Optional[Type[Any]] 

162 

163 if cls.__dict__.get("__abstract__", False): 

164 for base_ in cls.__bases__: 

165 sup = _resolve_for_abstract_or_classical(base_) 

166 if sup is not None: 

167 return sup 

168 else: 

169 return None 

170 else: 

171 clsmanager = _dive_for_cls_manager(cls) 

172 

173 if clsmanager: 

174 return clsmanager.class_ 

175 else: 

176 return cls 

177 

178 

179def _get_immediate_cls_attr( 

180 cls: Type[Any], attrname: str, strict: bool = False 

181) -> Optional[Any]: 

182 """return an attribute of the class that is either present directly 

183 on the class, e.g. not on a superclass, or is from a superclass but 

184 this superclass is a non-mapped mixin, that is, not a descendant of 

185 the declarative base and is also not classically mapped. 

186 

187 This is used to detect attributes that indicate something about 

188 a mapped class independently from any mapped classes that it may 

189 inherit from. 

190 

191 """ 

192 

193 # the rules are different for this name than others, 

194 # make sure we've moved it out. transitional 

195 assert attrname != "__abstract__" 

196 

197 if not issubclass(cls, object): 

198 return None 

199 

200 if attrname in cls.__dict__: 

201 return getattr(cls, attrname) 

202 

203 for base in cls.__mro__[1:]: 

204 _is_classical_inherits = _dive_for_cls_manager(base) is not None 

205 

206 if attrname in base.__dict__ and ( 

207 base is cls 

208 or ( 

209 (base in cls.__bases__ if strict else True) 

210 and not _is_classical_inherits 

211 ) 

212 ): 

213 return getattr(base, attrname) 

214 else: 

215 return None 

216 

217 

218def _dive_for_cls_manager(cls: Type[_O]) -> Optional[ClassManager[_O]]: 

219 # because the class manager registration is pluggable, 

220 # we need to do the search for every class in the hierarchy, 

221 # rather than just a simple "cls._sa_class_manager" 

222 

223 for base in cls.__mro__: 

224 manager: Optional[ClassManager[_O]] = attributes.opt_manager_of_class( 

225 base 

226 ) 

227 if manager: 

228 return manager 

229 return None 

230 

231 

232@util.preload_module("sqlalchemy.orm.decl_api") 

233def _is_declarative_props(obj: Any) -> bool: 

234 _declared_attr_common = util.preloaded.orm_decl_api._declared_attr_common 

235 

236 return isinstance(obj, (_declared_attr_common, util.classproperty)) 

237 

238 

239def _check_declared_props_nocascade( 

240 obj: Any, name: str, cls: Type[_O] 

241) -> bool: 

242 if _is_declarative_props(obj): 

243 if getattr(obj, "_cascading", False): 

244 util.warn( 

245 "@declared_attr.cascading is not supported on the %s " 

246 "attribute on class %s. This attribute invokes for " 

247 "subclasses in any case." % (name, cls) 

248 ) 

249 return True 

250 else: 

251 return False 

252 

253 

254class _ORMClassConfigurator: 

255 """Object that configures a class that's potentially going to be 

256 mapped, and/or turned into an ORM dataclass. 

257 

258 This is the base class for all the configurator objects. 

259 

260 """ 

261 

262 __slots__ = ("cls", "classname", "__weakref__") 

263 

264 cls: Type[Any] 

265 classname: str 

266 

267 def __init__(self, cls_: Type[Any]): 

268 self.cls = util.assert_arg_type(cls_, type, "cls_") 

269 self.classname = cls_.__name__ 

270 

271 @classmethod 

272 def _as_declarative( 

273 cls, registry: _RegistryType, cls_: Type[Any], dict_: _ClassDict 

274 ) -> Optional[_MapperConfig]: 

275 manager = attributes.opt_manager_of_class(cls_) 

276 if manager and manager.class_ is cls_: 

277 raise exc.InvalidRequestError( 

278 f"Class {cls_!r} already has been instrumented declaratively" 

279 ) 

280 

281 # allow subclassing an orm class with typed columns without 

282 # generating an orm class 

283 if cls_.__dict__.get("__abstract__", False) or issubclass( 

284 cls_, TypedColumns 

285 ): 

286 return None 

287 

288 defer_map = _get_immediate_cls_attr( 

289 cls_, "_sa_decl_prepare_nocascade", strict=True 

290 ) or hasattr(cls_, "_sa_decl_prepare") 

291 

292 if defer_map: 

293 return _DeferredDeclarativeConfig(registry, cls_, dict_) 

294 else: 

295 return _DeclarativeMapperConfig(registry, cls_, dict_) 

296 

297 @classmethod 

298 def _as_unmapped_dataclass( 

299 cls, cls_: Type[Any], dict_: _ClassDict 

300 ) -> _UnmappedDataclassConfig: 

301 return _UnmappedDataclassConfig(cls_, dict_) 

302 

303 @classmethod 

304 def _mapper( 

305 cls, 

306 registry: _RegistryType, 

307 cls_: Type[_O], 

308 table: Optional[FromClause], 

309 mapper_kw: _MapperKwArgs, 

310 ) -> Mapper[_O]: 

311 _ImperativeMapperConfig(registry, cls_, table, mapper_kw) 

312 return cast("MappedClassProtocol[_O]", cls_).__mapper__ 

313 

314 

315class _MapperConfig(_ORMClassConfigurator): 

316 """Configurator that configures a class that's potentially going to be 

317 mapped, and optionally turned into a dataclass as well.""" 

318 

319 __slots__ = ( 

320 "properties", 

321 "declared_attr_reg", 

322 ) 

323 

324 properties: util.OrderedDict[ 

325 str, 

326 Union[ 

327 Sequence[NamedColumn[Any]], NamedColumn[Any], MapperProperty[Any] 

328 ], 

329 ] 

330 declared_attr_reg: Dict[declared_attr[Any], Any] 

331 

332 def __init__( 

333 self, 

334 registry: _RegistryType, 

335 cls_: Type[Any], 

336 ): 

337 super().__init__(cls_) 

338 self.properties = util.OrderedDict() 

339 self.declared_attr_reg = {} 

340 

341 instrumentation.register_class( 

342 self.cls, 

343 finalize=False, 

344 registry=registry, 

345 declarative_scan=self, 

346 init_method=registry.constructor, 

347 ) 

348 

349 def set_cls_attribute(self, attrname: str, value: _T) -> _T: 

350 manager = instrumentation.manager_of_class(self.cls) 

351 manager.install_member(attrname, value) 

352 return value 

353 

354 def map(self, mapper_kw: _MapperKwArgs) -> Mapper[Any]: 

355 raise NotImplementedError() 

356 

357 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None: 

358 self.map(mapper_kw) 

359 

360 

361class _ImperativeMapperConfig(_MapperConfig): 

362 """Configurator that configures a class for an imperative mapping.""" 

363 

364 __slots__ = ("local_table", "inherits") 

365 

366 def __init__( 

367 self, 

368 registry: _RegistryType, 

369 cls_: Type[_O], 

370 table: Optional[FromClause], 

371 mapper_kw: _MapperKwArgs, 

372 ): 

373 super().__init__(registry, cls_) 

374 

375 self.local_table = self.set_cls_attribute("__table__", table) 

376 

377 with mapperlib._CONFIGURE_MUTEX: 

378 clsregistry._add_class( 

379 self.classname, self.cls, registry._class_registry 

380 ) 

381 

382 self._setup_inheritance(mapper_kw) 

383 

384 self._early_mapping(mapper_kw) 

385 

386 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]: 

387 mapper_cls = Mapper 

388 

389 return self.set_cls_attribute( 

390 "__mapper__", 

391 mapper_cls(self.cls, self.local_table, **mapper_kw), 

392 ) 

393 

394 def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None: 

395 cls = self.cls 

396 

397 inherits = None 

398 inherits_search = [] 

399 

400 # since we search for classical mappings now, search for 

401 # multiple mapped bases as well and raise an error. 

402 for base_ in cls.__bases__: 

403 c = _resolve_for_abstract_or_classical(base_) 

404 if c is None: 

405 continue 

406 

407 if _is_supercls_for_inherits(c) and c not in inherits_search: 

408 inherits_search.append(c) 

409 

410 if inherits_search: 

411 if len(inherits_search) > 1: 

412 raise exc.InvalidRequestError( 

413 "Class %s has multiple mapped bases: %r" 

414 % (cls, inherits_search) 

415 ) 

416 inherits = inherits_search[0] 

417 

418 self.inherits = inherits 

419 

420 

421class _CollectedAnnotation(NamedTuple): 

422 raw_annotation: _AnnotationScanType 

423 mapped_container: Optional[Type[Mapped[Any]]] 

424 extracted_mapped_annotation: Union[_AnnotationScanType, str] 

425 is_dataclass: bool 

426 attr_value: Any 

427 originating_module: str 

428 originating_class: Type[Any] 

429 

430 

431class _ClassScanAbstractConfig(_ORMClassConfigurator): 

432 """Abstract base for a configurator that configures a class for a 

433 declarative mapping, or an unmapped ORM dataclass. 

434 

435 Defines scanning of pep-484 annotations as well as ORM dataclass 

436 applicators 

437 

438 """ 

439 

440 __slots__ = () 

441 

442 clsdict_view: _ClassDict 

443 collected_annotations: Dict[str, _CollectedAnnotation] 

444 collected_attributes: Dict[str, Any] 

445 

446 is_dataclass_prior_to_mapping: bool 

447 allow_unmapped_annotations: bool 

448 

449 dataclass_setup_arguments: Optional[_DataclassArguments] 

450 """if the class has SQLAlchemy native dataclass parameters, where 

451 we will turn the class into a dataclass within the declarative mapping 

452 process. 

453 

454 """ 

455 

456 allow_dataclass_fields: bool 

457 """if true, look for dataclass-processed Field objects on the target 

458 class as well as superclasses and extract ORM mapping directives from 

459 the "metadata" attribute of each Field. 

460 

461 if False, dataclass fields can still be used, however they won't be 

462 mapped. 

463 

464 """ 

465 

466 _include_dunders = { 

467 "__table__", 

468 "__mapper_args__", 

469 "__tablename__", 

470 "__table_args__", 

471 } 

472 

473 _match_exclude_dunders = re.compile(r"^(?:_sa_|__)") 

474 

475 def _scan_attributes(self) -> None: 

476 raise NotImplementedError() 

477 

478 def _setup_dataclasses_transforms( 

479 self, *, enable_descriptor_defaults: bool, revert: bool = False 

480 ) -> None: 

481 dataclass_setup_arguments = self.dataclass_setup_arguments 

482 if not dataclass_setup_arguments: 

483 return 

484 

485 # can't use is_dataclass since it uses hasattr 

486 if "__dataclass_fields__" in self.cls.__dict__: 

487 raise exc.InvalidRequestError( 

488 f"Class {self.cls} is already a dataclass; ensure that " 

489 "base classes / decorator styles of establishing dataclasses " 

490 "are not being mixed. " 

491 "This can happen if a class that inherits from " 

492 "'MappedAsDataclass', even indirectly, is been mapped with " 

493 "'@registry.mapped_as_dataclass'" 

494 ) 

495 

496 # can't create a dataclass if __table__ is already there. This would 

497 # fail an assertion when calling _get_arguments_for_make_dataclass: 

498 # assert False, "Mapped[] received without a mapping declaration" 

499 if "__table__" in self.cls.__dict__: 

500 raise exc.InvalidRequestError( 

501 f"Class {self.cls} already defines a '__table__'. " 

502 "ORM Annotated Dataclasses do not support a pre-existing " 

503 "'__table__' element" 

504 ) 

505 

506 raise_for_non_dc_attrs = collections.defaultdict(list) 

507 

508 def _allow_dataclass_field( 

509 key: str, originating_class: Type[Any] 

510 ) -> bool: 

511 if ( 

512 originating_class is not self.cls 

513 and "__dataclass_fields__" not in originating_class.__dict__ 

514 ): 

515 raise_for_non_dc_attrs[originating_class].append(key) 

516 

517 return True 

518 

519 field_list = [ 

520 _AttributeOptions._get_arguments_for_make_dataclass( 

521 self, 

522 key, 

523 anno, 

524 mapped_container, 

525 self.collected_attributes.get(key, _NoArg.NO_ARG), 

526 dataclass_setup_arguments, 

527 enable_descriptor_defaults, 

528 ) 

529 for key, anno, mapped_container in ( 

530 ( 

531 key, 

532 raw_anno, 

533 mapped_container, 

534 ) 

535 for key, ( 

536 raw_anno, 

537 mapped_container, 

538 mapped_anno, 

539 is_dc, 

540 attr_value, 

541 originating_module, 

542 originating_class, 

543 ) in self.collected_annotations.items() 

544 if _allow_dataclass_field(key, originating_class) 

545 and ( 

546 key not in self.collected_attributes 

547 # issue #9226; check for attributes that we've collected 

548 # which are already instrumented, which we would assume 

549 # mean we are in an ORM inheritance mapping and this 

550 # attribute is already mapped on the superclass. Under 

551 # no circumstance should any QueryableAttribute be sent to 

552 # the dataclass() function; anything that's mapped should 

553 # be Field and that's it 

554 or not isinstance( 

555 self.collected_attributes[key], QueryableAttribute 

556 ) 

557 ) 

558 ) 

559 ] 

560 if raise_for_non_dc_attrs: 

561 for ( 

562 originating_class, 

563 non_dc_attrs, 

564 ) in raise_for_non_dc_attrs.items(): 

565 raise exc.InvalidRequestError( 

566 f"When transforming {self.cls} to a dataclass, " 

567 f"attribute(s) " 

568 f"{', '.join(repr(key) for key in non_dc_attrs)} " 

569 f"originates from superclass " 

570 f"{originating_class}, which is not a dataclass. When " 

571 f"declaring SQLAlchemy Declarative " 

572 f"Dataclasses, ensure that all mixin classes and other " 

573 f"superclasses which include attributes are also a " 

574 f"subclass of MappedAsDataclass or make use of the " 

575 f"@unmapped_dataclass decorator.", 

576 code="dcmx", 

577 ) 

578 

579 if revert: 

580 # the "revert" case is used only by an unmapped mixin class 

581 # that is nonetheless using Mapped construct and needs to 

582 # itself be a dataclass 

583 revert_dict = { 

584 name: self.cls.__dict__[name] 

585 for name in (item[0] for item in field_list) 

586 if name in self.cls.__dict__ 

587 } 

588 else: 

589 revert_dict = None 

590 

591 # get original annotations using ForwardRef for symbols that 

592 # are unresolvable 

593 orig_annotations = util.get_annotations(self.cls) 

594 

595 # build a new __annotations__ dict from the fields we have. 

596 # this has to be done carefully since we have to maintain 

597 # the correct order! wow 

598 swap_annotations = {} 

599 defaults = {} 

600 

601 for item in field_list: 

602 if len(item) == 2: 

603 name, tp = item 

604 elif len(item) == 3: 

605 name, tp, spec = item 

606 defaults[name] = spec 

607 else: 

608 assert False 

609 

610 # add the annotation to the new dict we are creating. 

611 # note that if name is in orig_annotations, we expect 

612 # tp and orig_annotations[name] to be identical. 

613 swap_annotations[name] = orig_annotations.get(name, tp) 

614 

615 for k, v in defaults.items(): 

616 setattr(self.cls, k, v) 

617 

618 self._assert_dc_arguments(dataclass_setup_arguments) 

619 

620 dataclass_callable = dataclass_setup_arguments["dataclass_callable"] 

621 if dataclass_callable is _NoArg.NO_ARG: 

622 dataclass_callable = dataclasses.dataclass 

623 

624 # create a merged __annotations__ dictionary, maintaining order 

625 # as best we can: 

626 

627 # 1. merge all keys in orig_annotations that occur before 

628 # we see any of our mapped fields (this can be attributes like 

629 # __table_args__ etc.) 

630 new_annotations = { 

631 k: orig_annotations[k] 

632 for k in itertools.takewhile( 

633 lambda k: k not in swap_annotations, orig_annotations 

634 ) 

635 } 

636 

637 # 2. then put in all the dataclass annotations we have 

638 new_annotations |= swap_annotations 

639 

640 # 3. them merge all of orig_annotations which will add remaining 

641 # keys 

642 new_annotations |= orig_annotations 

643 

644 # 4. this becomes the new class annotations. 

645 restore_anno = util.restore_annotations(self.cls, new_annotations) 

646 

647 try: 

648 dataclass_callable( # type: ignore[call-overload] 

649 self.cls, 

650 **{ # type: ignore[call-overload,unused-ignore] 

651 k: v 

652 for k, v in dataclass_setup_arguments.items() 

653 if v is not _NoArg.NO_ARG 

654 and k not in ("dataclass_callable",) 

655 }, 

656 ) 

657 except (TypeError, ValueError) as ex: 

658 raise exc.InvalidRequestError( 

659 f"Python dataclasses error encountered when creating " 

660 f"dataclass for {self.cls.__name__!r}: " 

661 f"{ex!r}. Please refer to Python dataclasses " 

662 "documentation for additional information.", 

663 code="dcte", 

664 ) from ex 

665 else: 

666 # as of Python 3.15, dataclasses no longer renders the 

667 # auto-generated class docstring immediately; it instead installs 

668 # a descriptor that renders the ``__init__`` signature the first 

669 # time ``__doc__`` is accessed (see 

670 # ``dataclasses._AutoDocstring``). As the ``finally:`` block 

671 # below puts the class' original annotations back, that deferred 

672 # render would no longer see the dataclass-oriented annotations 

673 # we swapped in above, and would omit them from the docstring 

674 # entirely. Read the attribute now, while those annotations are 

675 # still in place, so the docstring we generate is the same on 

676 # every Python version. 

677 self.cls.__doc__ 

678 

679 finally: 

680 if revert and revert_dict: 

681 # used for mixin dataclasses; we have to restore the 

682 # mapped_column(), relationship() etc. to the class so these 

683 # take place for a mapped class scan 

684 for k, v in revert_dict.items(): 

685 setattr(self.cls, k, v) 

686 

687 restore_anno() 

688 

689 def _collect_annotation( 

690 self, 

691 name: str, 

692 raw_annotation: _AnnotationScanType, 

693 originating_class: Type[Any], 

694 expect_mapped: Optional[bool], 

695 attr_value: Any, 

696 ) -> Optional[_CollectedAnnotation]: 

697 if name in self.collected_annotations: 

698 return self.collected_annotations[name] 

699 

700 if raw_annotation is None: 

701 return None 

702 

703 is_dataclass = self.is_dataclass_prior_to_mapping 

704 allow_unmapped = self.allow_unmapped_annotations 

705 

706 if expect_mapped is None: 

707 is_dataclass_field = isinstance(attr_value, dataclasses.Field) 

708 expect_mapped = ( 

709 not is_dataclass_field 

710 and not allow_unmapped 

711 and ( 

712 attr_value is None 

713 or isinstance(attr_value, _MappedAttribute) 

714 ) 

715 ) 

716 

717 is_dataclass_field = False 

718 extracted = _extract_mapped_subtype( 

719 raw_annotation, 

720 self.cls, 

721 originating_class.__module__, 

722 name, 

723 type(attr_value), 

724 required=False, 

725 is_dataclass_field=is_dataclass_field, 

726 expect_mapped=expect_mapped and not is_dataclass, 

727 ) 

728 if extracted is None: 

729 # ClassVar can come out here 

730 return None 

731 

732 extracted_mapped_annotation, mapped_container = extracted 

733 

734 if attr_value is None and not is_literal(extracted_mapped_annotation): 

735 for elem in get_args(extracted_mapped_annotation): 

736 if is_fwd_ref( 

737 elem, check_generic=True, check_for_plain_string=True 

738 ): 

739 elem = de_stringify_annotation( 

740 self.cls, 

741 elem, 

742 originating_class.__module__, 

743 include_generic=True, 

744 ) 

745 # look in Annotated[...] for an ORM construct, 

746 # such as Annotated[int, mapped_column(primary_key=True)] 

747 if isinstance(elem, _IntrospectsAnnotations): 

748 attr_value = elem.found_in_pep593_annotated() 

749 

750 self.collected_annotations[name] = ca = _CollectedAnnotation( 

751 raw_annotation, 

752 mapped_container, 

753 extracted_mapped_annotation, 

754 is_dataclass, 

755 attr_value, 

756 originating_class.__module__, 

757 originating_class, 

758 ) 

759 return ca 

760 

761 @classmethod 

762 def _assert_dc_arguments(cls, arguments: _DataclassArguments) -> None: 

763 allowed = { 

764 "init", 

765 "repr", 

766 "order", 

767 "eq", 

768 "unsafe_hash", 

769 "kw_only", 

770 "match_args", 

771 "dataclass_callable", 

772 } 

773 disallowed_args = set(arguments).difference(allowed) 

774 if disallowed_args: 

775 msg = ", ".join(f"{arg!r}" for arg in sorted(disallowed_args)) 

776 raise exc.ArgumentError( 

777 f"Dataclass argument(s) {msg} are not accepted" 

778 ) 

779 

780 def _cls_attr_override_checker( 

781 self, cls: Type[_O] 

782 ) -> Callable[[str, Any], bool]: 

783 """Produce a function that checks if a class has overridden an 

784 attribute, taking SQLAlchemy-enabled dataclass fields into account. 

785 

786 """ 

787 

788 if self.allow_dataclass_fields: 

789 sa_dataclass_metadata_key = _get_immediate_cls_attr( 

790 cls, "__sa_dataclass_metadata_key__" 

791 ) 

792 else: 

793 sa_dataclass_metadata_key = None 

794 

795 if not sa_dataclass_metadata_key: 

796 

797 def attribute_is_overridden(key: str, obj: Any) -> bool: 

798 return getattr(cls, key, obj) is not obj 

799 

800 else: 

801 all_datacls_fields = { 

802 f.name: f.metadata[sa_dataclass_metadata_key] 

803 for f in util.dataclass_fields(cls) 

804 if sa_dataclass_metadata_key in f.metadata 

805 } 

806 local_datacls_fields = { 

807 f.name: f.metadata[sa_dataclass_metadata_key] 

808 for f in util.local_dataclass_fields(cls) 

809 if sa_dataclass_metadata_key in f.metadata 

810 } 

811 

812 absent = object() 

813 

814 def attribute_is_overridden(key: str, obj: Any) -> bool: 

815 if _is_declarative_props(obj): 

816 obj = obj.fget 

817 

818 # this function likely has some failure modes still if 

819 # someone is doing a deep mixing of the same attribute 

820 # name as plain Python attribute vs. dataclass field. 

821 

822 ret = local_datacls_fields.get(key, absent) 

823 if _is_declarative_props(ret): 

824 ret = ret.fget 

825 

826 if ret is obj: 

827 return False 

828 elif ret is not absent: 

829 return True 

830 

831 all_field = all_datacls_fields.get(key, absent) 

832 

833 ret = getattr(cls, key, obj) 

834 

835 if ret is obj: 

836 return False 

837 

838 # for dataclasses, this could be the 

839 # 'default' of the field. so filter more specifically 

840 # for an already-mapped InstrumentedAttribute 

841 if ret is not absent and isinstance( 

842 ret, InstrumentedAttribute 

843 ): 

844 return True 

845 

846 if all_field is obj: 

847 return False 

848 elif all_field is not absent: 

849 return True 

850 

851 # can't find another attribute 

852 return False 

853 

854 return attribute_is_overridden 

855 

856 def _cls_attr_resolver( 

857 self, cls: Type[Any] 

858 ) -> Callable[[], Iterable[Tuple[str, Any, Any, bool]]]: 

859 """produce a function to iterate the "attributes" of a class 

860 which we want to consider for mapping, adjusting for SQLAlchemy fields 

861 embedded in dataclass fields. 

862 

863 """ 

864 cls_annotations = util.get_annotations(cls) 

865 

866 cls_vars = vars(cls) 

867 

868 _include_dunders = self._include_dunders 

869 _match_exclude_dunders = self._match_exclude_dunders 

870 

871 names = [ 

872 n 

873 for n in util.merge_lists_w_ordering( 

874 list(cls_vars), list(cls_annotations) 

875 ) 

876 if not _match_exclude_dunders.match(n) or n in _include_dunders 

877 ] 

878 

879 if self.allow_dataclass_fields: 

880 sa_dataclass_metadata_key: Optional[str] = _get_immediate_cls_attr( 

881 cls, "__sa_dataclass_metadata_key__" 

882 ) 

883 else: 

884 sa_dataclass_metadata_key = None 

885 

886 if not sa_dataclass_metadata_key: 

887 

888 def local_attributes_for_class() -> ( 

889 Iterable[Tuple[str, Any, Any, bool]] 

890 ): 

891 return ( 

892 ( 

893 name, 

894 cls_vars.get(name), 

895 cls_annotations.get(name), 

896 False, 

897 ) 

898 for name in names 

899 ) 

900 

901 else: 

902 dataclass_fields = { 

903 field.name: field for field in util.local_dataclass_fields(cls) 

904 } 

905 

906 fixed_sa_dataclass_metadata_key = sa_dataclass_metadata_key 

907 

908 def local_attributes_for_class() -> ( 

909 Iterable[Tuple[str, Any, Any, bool]] 

910 ): 

911 for name in names: 

912 field = dataclass_fields.get(name, None) 

913 if field and sa_dataclass_metadata_key in field.metadata: 

914 yield field.name, _as_dc_declaredattr( 

915 field.metadata, fixed_sa_dataclass_metadata_key 

916 ), cls_annotations.get(field.name), True 

917 else: 

918 yield name, cls_vars.get(name), cls_annotations.get( 

919 name 

920 ), False 

921 

922 return local_attributes_for_class 

923 

924 

925class _DeclarativeMapperConfig(_MapperConfig, _ClassScanAbstractConfig): 

926 """Configurator that will produce a declarative mapped class""" 

927 

928 __slots__ = ( 

929 "registry", 

930 "local_table", 

931 "persist_selectable", 

932 "declared_columns", 

933 "column_ordering", 

934 "column_copies", 

935 "table_args", 

936 "tablename", 

937 "mapper_args", 

938 "mapper_args_fn", 

939 "table_fn", 

940 "inherits", 

941 "single", 

942 "clsdict_view", 

943 "collected_attributes", 

944 "collected_annotations", 

945 "allow_dataclass_fields", 

946 "dataclass_setup_arguments", 

947 "is_dataclass_prior_to_mapping", 

948 "allow_unmapped_annotations", 

949 ) 

950 

951 is_deferred = False 

952 registry: _RegistryType 

953 local_table: Optional[FromClause] 

954 persist_selectable: Optional[FromClause] 

955 declared_columns: util.OrderedSet[Column[Any]] 

956 column_ordering: Dict[Column[Any], int] 

957 column_copies: Dict[ 

958 Union[MappedColumn[Any], Column[Any]], 

959 Union[MappedColumn[Any], Column[Any]], 

960 ] 

961 tablename: Optional[str] 

962 mapper_args: Mapping[str, Any] 

963 table_args: Optional[_TableArgsType] 

964 mapper_args_fn: Optional[Callable[[], Dict[str, Any]]] 

965 inherits: Optional[Type[Any]] 

966 single: bool 

967 

968 def __init__( 

969 self, 

970 registry: _RegistryType, 

971 cls_: Type[_O], 

972 dict_: _ClassDict, 

973 ): 

974 # grab class dict before the instrumentation manager has been added. 

975 # reduces cycles 

976 self.clsdict_view = ( 

977 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT 

978 ) 

979 super().__init__(registry, cls_) 

980 self.registry = registry 

981 self.persist_selectable = None 

982 

983 self.collected_attributes = {} 

984 self.collected_annotations = {} 

985 self.declared_columns = util.OrderedSet() 

986 self.column_ordering = {} 

987 self.column_copies = {} 

988 self.single = False 

989 self.dataclass_setup_arguments = dca = getattr( 

990 self.cls, "_sa_apply_dc_transforms", None 

991 ) 

992 

993 self.allow_unmapped_annotations = getattr( 

994 self.cls, "__allow_unmapped__", False 

995 ) or bool(self.dataclass_setup_arguments) 

996 

997 self.is_dataclass_prior_to_mapping = cld = dataclasses.is_dataclass( 

998 cls_ 

999 ) 

1000 

1001 sdk = _get_immediate_cls_attr(cls_, "__sa_dataclass_metadata_key__") 

1002 

1003 # we don't want to consume Field objects from a not-already-dataclass. 

1004 # the Field objects won't have their "name" or "type" populated, 

1005 # and while it seems like we could just set these on Field as we 

1006 # read them, Field is documented as "user read only" and we need to 

1007 # stay far away from any off-label use of dataclasses APIs. 

1008 if (not cld or dca) and sdk: 

1009 raise exc.InvalidRequestError( 

1010 "SQLAlchemy mapped dataclasses can't consume mapping " 

1011 "information from dataclass.Field() objects if the immediate " 

1012 "class is not already a dataclass." 

1013 ) 

1014 

1015 # if already a dataclass, and __sa_dataclass_metadata_key__ present, 

1016 # then also look inside of dataclass.Field() objects yielded by 

1017 # dataclasses.get_fields(cls) when scanning for attributes 

1018 self.allow_dataclass_fields = bool(sdk and cld) 

1019 

1020 self._setup_declared_events() 

1021 

1022 self._scan_attributes() 

1023 

1024 self._setup_dataclasses_transforms(enable_descriptor_defaults=True) 

1025 

1026 with mapperlib._CONFIGURE_MUTEX: 

1027 clsregistry._add_class( 

1028 self.classname, self.cls, registry._class_registry 

1029 ) 

1030 

1031 self._setup_inheriting_mapper() 

1032 

1033 self._extract_mappable_attributes() 

1034 

1035 self._extract_declared_columns() 

1036 

1037 self._setup_table() 

1038 

1039 self._setup_inheriting_columns() 

1040 

1041 self._early_mapping(util.EMPTY_DICT) 

1042 

1043 def _setup_declared_events(self) -> None: 

1044 if _get_immediate_cls_attr(self.cls, "__declare_last__"): 

1045 

1046 @event.listens_for(Mapper, "after_configured") 

1047 def after_configured() -> None: 

1048 cast( 

1049 "_DeclMappedClassProtocol[Any]", self.cls 

1050 ).__declare_last__() 

1051 

1052 if _get_immediate_cls_attr(self.cls, "__declare_first__"): 

1053 

1054 @event.listens_for(Mapper, "before_configured") 

1055 def before_configured() -> None: 

1056 cast( 

1057 "_DeclMappedClassProtocol[Any]", self.cls 

1058 ).__declare_first__() 

1059 

1060 def _scan_attributes(self) -> None: 

1061 cls = self.cls 

1062 

1063 cls_as_Decl = cast("_DeclMappedClassProtocol[Any]", cls) 

1064 

1065 clsdict_view = self.clsdict_view 

1066 collected_attributes = self.collected_attributes 

1067 column_copies = self.column_copies 

1068 _include_dunders = self._include_dunders 

1069 mapper_args_fn = None 

1070 table_args = inherited_table_args = None 

1071 table_fn = None 

1072 tablename = None 

1073 fixed_table = "__table__" in clsdict_view 

1074 

1075 attribute_is_overridden = self._cls_attr_override_checker(self.cls) 

1076 

1077 bases = [] 

1078 

1079 for base in cls.__mro__: 

1080 # collect bases and make sure standalone columns are copied 

1081 # to be the column they will ultimately be on the class, 

1082 # so that declared_attr functions use the right columns. 

1083 # need to do this all the way up the hierarchy first 

1084 # (see #8190) 

1085 

1086 class_mapped = base is not cls and _is_supercls_for_inherits(base) 

1087 

1088 local_attributes_for_class = self._cls_attr_resolver(base) 

1089 

1090 if not class_mapped and base is not cls: 

1091 locally_collected_columns = self._produce_column_copies( 

1092 local_attributes_for_class, 

1093 attribute_is_overridden, 

1094 fixed_table, 

1095 base, 

1096 ) 

1097 else: 

1098 locally_collected_columns = {} 

1099 

1100 bases.append( 

1101 ( 

1102 base, 

1103 class_mapped, 

1104 local_attributes_for_class, 

1105 locally_collected_columns, 

1106 ) 

1107 ) 

1108 

1109 for ( 

1110 base, 

1111 class_mapped, 

1112 local_attributes_for_class, 

1113 locally_collected_columns, 

1114 ) in bases: 

1115 # this transfer can also take place as we scan each name 

1116 # for finer-grained control of how collected_attributes is 

1117 # populated, as this is what impacts column ordering. 

1118 # however it's simpler to get it out of the way here. 

1119 collected_attributes.update(locally_collected_columns) 

1120 

1121 for ( 

1122 name, 

1123 obj, 

1124 annotation, 

1125 is_dataclass_field, 

1126 ) in local_attributes_for_class(): 

1127 if name in _include_dunders: 

1128 if name == "__mapper_args__": 

1129 check_decl = _check_declared_props_nocascade( 

1130 obj, name, cls 

1131 ) 

1132 if not mapper_args_fn and ( 

1133 not class_mapped or check_decl 

1134 ): 

1135 # don't even invoke __mapper_args__ until 

1136 # after we've determined everything about the 

1137 # mapped table. 

1138 # make a copy of it so a class-level dictionary 

1139 # is not overwritten when we update column-based 

1140 # arguments. 

1141 def _mapper_args_fn() -> Dict[str, Any]: 

1142 return dict(cls_as_Decl.__mapper_args__) 

1143 

1144 mapper_args_fn = _mapper_args_fn 

1145 

1146 elif name == "__tablename__": 

1147 check_decl = _check_declared_props_nocascade( 

1148 obj, name, cls 

1149 ) 

1150 if not tablename and (not class_mapped or check_decl): 

1151 tablename = cls_as_Decl.__tablename__ 

1152 elif name == "__table__": 

1153 check_decl = _check_declared_props_nocascade( 

1154 obj, name, cls 

1155 ) 

1156 # if a @declared_attr using "__table__" is detected, 

1157 # wrap up a callable to look for "__table__" from 

1158 # the final concrete class when we set up a table. 

1159 # this was fixed by 

1160 # #11509, regression in 2.0 from version 1.4. 

1161 if check_decl and not table_fn: 

1162 # don't even invoke __table__ until we're ready 

1163 def _table_fn() -> FromClause: 

1164 return cls_as_Decl.__table__ 

1165 

1166 table_fn = _table_fn 

1167 

1168 elif name == "__table_args__": 

1169 check_decl = _check_declared_props_nocascade( 

1170 obj, name, cls 

1171 ) 

1172 if not table_args and (not class_mapped or check_decl): 

1173 table_args = cls_as_Decl.__table_args__ 

1174 if not isinstance( 

1175 table_args, (tuple, dict, type(None)) 

1176 ): 

1177 raise exc.ArgumentError( 

1178 "__table_args__ value must be a tuple, " 

1179 "dict, or None" 

1180 ) 

1181 if base is not cls: 

1182 inherited_table_args = True 

1183 else: 

1184 # any other dunder names; should not be here 

1185 # as we have tested for all four names in 

1186 # _include_dunders 

1187 assert False 

1188 elif class_mapped: 

1189 if _is_declarative_props(obj) and not obj._quiet: 

1190 util.warn( 

1191 "Regular (i.e. not __special__) " 

1192 "attribute '%s.%s' uses @declared_attr, " 

1193 "but owning class %s is mapped - " 

1194 "not applying to subclass %s." 

1195 % (base.__name__, name, base, cls) 

1196 ) 

1197 

1198 continue 

1199 elif base is not cls: 

1200 # we're a mixin, abstract base, or something that is 

1201 # acting like that for now. 

1202 

1203 if isinstance(obj, (Column, MappedColumn)): 

1204 # already copied columns to the mapped class. 

1205 continue 

1206 elif isinstance(obj, MapperProperty): 

1207 raise exc.InvalidRequestError( 

1208 "Mapper properties (i.e. deferred," 

1209 "column_property(), relationship(), etc.) must " 

1210 "be declared as @declared_attr callables " 

1211 "on declarative mixin classes. For dataclass " 

1212 "field() objects, use a lambda:" 

1213 ) 

1214 elif _is_declarative_props(obj): 

1215 # tried to get overloads to tell this to 

1216 # pylance, no luck 

1217 assert obj is not None 

1218 

1219 if obj._cascading: 

1220 if name in clsdict_view: 

1221 # unfortunately, while we can use the user- 

1222 # defined attribute here to allow a clean 

1223 # override, if there's another 

1224 # subclass below then it still tries to use 

1225 # this. not sure if there is enough 

1226 # information here to add this as a feature 

1227 # later on. 

1228 util.warn( 

1229 "Attribute '%s' on class %s cannot be " 

1230 "processed due to " 

1231 "@declared_attr.cascading; " 

1232 "skipping" % (name, cls) 

1233 ) 

1234 collected_attributes[name] = column_copies[obj] = ( 

1235 ret 

1236 ) = obj.__get__(obj, cls) 

1237 setattr(cls, name, ret) 

1238 else: 

1239 if is_dataclass_field: 

1240 # access attribute using normal class access 

1241 # first, to see if it's been mapped on a 

1242 # superclass. note if the dataclasses.field() 

1243 # has "default", this value can be anything. 

1244 ret = getattr(cls, name, None) 

1245 

1246 # so, if it's anything that's not ORM 

1247 # mapped, assume we should invoke the 

1248 # declared_attr 

1249 if not isinstance(ret, InspectionAttr): 

1250 ret = obj.fget() 

1251 else: 

1252 # access attribute using normal class access. 

1253 # if the declared attr already took place 

1254 # on a superclass that is mapped, then 

1255 # this is no longer a declared_attr, it will 

1256 # be the InstrumentedAttribute 

1257 ret = getattr(cls, name) 

1258 

1259 # correct for proxies created from hybrid_property 

1260 # or similar. note there is no known case that 

1261 # produces nested proxies, so we are only 

1262 # looking one level deep right now. 

1263 

1264 if ( 

1265 isinstance(ret, InspectionAttr) 

1266 and attr_is_internal_proxy(ret) 

1267 and not isinstance( 

1268 ret.original_property, MapperProperty 

1269 ) 

1270 ): 

1271 ret = ret.descriptor 

1272 

1273 collected_attributes[name] = column_copies[obj] = ( 

1274 ret 

1275 ) 

1276 

1277 if ( 

1278 isinstance(ret, (Column, MapperProperty)) 

1279 and ret.doc is None 

1280 ): 

1281 ret.doc = obj.__doc__ 

1282 

1283 self._collect_annotation( 

1284 name, 

1285 obj._collect_return_annotation(), 

1286 base, 

1287 True, 

1288 obj, 

1289 ) 

1290 elif _is_mapped_annotation(annotation, cls, base): 

1291 # Mapped annotation without any object. 

1292 # product_column_copies should have handled this. 

1293 # if future support for other MapperProperty, 

1294 # then test if this name is already handled and 

1295 # otherwise proceed to generate. 

1296 if not fixed_table: 

1297 assert ( 

1298 name in collected_attributes 

1299 or attribute_is_overridden(name, None) 

1300 ) 

1301 continue 

1302 else: 

1303 # here, the attribute is some other kind of 

1304 # property that we assume is not part of the 

1305 # declarative mapping. however, check for some 

1306 # more common mistakes 

1307 self._warn_for_decl_attributes(base, name, obj) 

1308 elif is_dataclass_field and ( 

1309 name not in clsdict_view or clsdict_view[name] is not obj 

1310 ): 

1311 # here, we are definitely looking at the target class 

1312 # and not a superclass. this is currently a 

1313 # dataclass-only path. if the name is only 

1314 # a dataclass field and isn't in local cls.__dict__, 

1315 # put the object there. 

1316 # assert that the dataclass-enabled resolver agrees 

1317 # with what we are seeing 

1318 

1319 assert not attribute_is_overridden(name, obj) 

1320 

1321 if _is_declarative_props(obj): 

1322 obj = obj.fget() 

1323 

1324 collected_attributes[name] = obj 

1325 self._collect_annotation( 

1326 name, annotation, base, False, obj 

1327 ) 

1328 else: 

1329 collected_annotation = self._collect_annotation( 

1330 name, annotation, base, None, obj 

1331 ) 

1332 is_mapped = ( 

1333 collected_annotation is not None 

1334 and collected_annotation.mapped_container is not None 

1335 ) 

1336 generated_obj = ( 

1337 collected_annotation.attr_value 

1338 if collected_annotation is not None 

1339 else obj 

1340 ) 

1341 if obj is None and not fixed_table and is_mapped: 

1342 collected_attributes[name] = ( 

1343 generated_obj 

1344 if generated_obj is not None 

1345 else MappedColumn() 

1346 ) 

1347 elif name in clsdict_view: 

1348 collected_attributes[name] = obj 

1349 # else if the name is not in the cls.__dict__, 

1350 # don't collect it as an attribute. 

1351 # we will see the annotation only, which is meaningful 

1352 # both for mapping and dataclasses setup 

1353 

1354 if inherited_table_args and not tablename: 

1355 table_args = None 

1356 

1357 self.table_args = table_args 

1358 self.tablename = tablename 

1359 self.mapper_args_fn = mapper_args_fn 

1360 self.table_fn = table_fn 

1361 

1362 @classmethod 

1363 def _update_annotations_for_non_mapped_class( 

1364 cls, klass: Type[_O] 

1365 ) -> Mapping[str, _AnnotationScanType]: 

1366 cls_annotations = util.get_annotations(klass) 

1367 

1368 new_anno = {} 

1369 for name, annotation in cls_annotations.items(): 

1370 if _is_mapped_annotation(annotation, klass, klass): 

1371 extracted = _extract_mapped_subtype( 

1372 annotation, 

1373 klass, 

1374 klass.__module__, 

1375 name, 

1376 type(None), 

1377 required=False, 

1378 is_dataclass_field=False, 

1379 expect_mapped=False, 

1380 ) 

1381 if extracted: 

1382 inner, _ = extracted 

1383 new_anno[name] = inner 

1384 else: 

1385 new_anno[name] = annotation 

1386 return new_anno 

1387 

1388 def _warn_for_decl_attributes( 

1389 self, cls: Type[Any], key: str, c: Any 

1390 ) -> None: 

1391 if isinstance(c, expression.ColumnElement): 

1392 util.warn( 

1393 f"Attribute '{key}' on class {cls} appears to " 

1394 "be a non-schema SQLAlchemy expression " 

1395 "object; this won't be part of the declarative mapping. " 

1396 "To map arbitrary expressions, use ``column_property()`` " 

1397 "or a similar function such as ``deferred()``, " 

1398 "``query_expression()`` etc. " 

1399 ) 

1400 

1401 def _produce_column_copies( 

1402 self, 

1403 attributes_for_class: Callable[ 

1404 [], Iterable[Tuple[str, Any, Any, bool]] 

1405 ], 

1406 attribute_is_overridden: Callable[[str, Any], bool], 

1407 fixed_table: bool, 

1408 originating_class: Type[Any], 

1409 ) -> Dict[str, Union[Column[Any], MappedColumn[Any]]]: 

1410 cls = self.cls 

1411 dict_ = self.clsdict_view 

1412 locally_collected_attributes = {} 

1413 column_copies = self.column_copies 

1414 # copy mixin columns to the mapped class 

1415 

1416 for name, obj, annotation, is_dataclass in attributes_for_class(): 

1417 if ( 

1418 not fixed_table 

1419 and obj is None 

1420 and _is_mapped_annotation(annotation, cls, originating_class) 

1421 ): 

1422 # obj is None means this is the annotation only path 

1423 

1424 if attribute_is_overridden(name, obj): 

1425 # perform same "overridden" check as we do for 

1426 # Column/MappedColumn, this is how a mixin col is not 

1427 # applied to an inherited subclass that does not have 

1428 # the mixin. the anno-only path added here for 

1429 # #9564 

1430 continue 

1431 

1432 collected_annotation = self._collect_annotation( 

1433 name, annotation, originating_class, True, obj 

1434 ) 

1435 obj = ( 

1436 collected_annotation.attr_value 

1437 if collected_annotation is not None 

1438 else obj 

1439 ) 

1440 if obj is None: 

1441 obj = MappedColumn() 

1442 

1443 locally_collected_attributes[name] = obj 

1444 setattr(cls, name, obj) 

1445 

1446 elif isinstance(obj, (Column, MappedColumn)): 

1447 if attribute_is_overridden(name, obj): 

1448 # if column has been overridden 

1449 # (like by the InstrumentedAttribute of the 

1450 # superclass), skip. don't collect the annotation 

1451 # either (issue #8718) 

1452 continue 

1453 

1454 collected_annotation = self._collect_annotation( 

1455 name, annotation, originating_class, True, obj 

1456 ) 

1457 obj = ( 

1458 collected_annotation.attr_value 

1459 if collected_annotation is not None 

1460 else obj 

1461 ) 

1462 

1463 if name not in dict_ and not ( 

1464 "__table__" in dict_ 

1465 and (getattr(obj, "name", None) or name) 

1466 in dict_["__table__"].c 

1467 ): 

1468 if obj.foreign_keys: 

1469 for fk in obj.foreign_keys: 

1470 if ( 

1471 fk._table_column is not None 

1472 and fk._table_column.table is None 

1473 ): 

1474 raise exc.InvalidRequestError( 

1475 "Columns with foreign keys to " 

1476 "non-table-bound " 

1477 "columns must be declared as " 

1478 "@declared_attr callables " 

1479 "on declarative mixin classes. " 

1480 "For dataclass " 

1481 "field() objects, use a lambda:." 

1482 ) 

1483 

1484 column_copies[obj] = copy_ = obj._copy() 

1485 

1486 locally_collected_attributes[name] = copy_ 

1487 setattr(cls, name, copy_) 

1488 

1489 return locally_collected_attributes 

1490 

1491 def _extract_mappable_attributes(self) -> None: 

1492 cls = self.cls 

1493 collected_attributes = self.collected_attributes 

1494 

1495 our_stuff = self.properties 

1496 

1497 _include_dunders = self._include_dunders 

1498 

1499 late_mapped = _get_immediate_cls_attr( 

1500 cls, "_sa_decl_prepare_nocascade", strict=True 

1501 ) 

1502 

1503 allow_unmapped_annotations = self.allow_unmapped_annotations 

1504 expect_annotations_wo_mapped = ( 

1505 allow_unmapped_annotations or self.is_dataclass_prior_to_mapping 

1506 ) 

1507 

1508 look_for_dataclass_things = bool(self.dataclass_setup_arguments) 

1509 

1510 for k in list(collected_attributes): 

1511 if k in _include_dunders: 

1512 continue 

1513 

1514 value = collected_attributes[k] 

1515 

1516 if _is_declarative_props(value): 

1517 # @declared_attr in collected_attributes only occurs here for a 

1518 # @declared_attr that's directly on the mapped class; 

1519 # for a mixin, these have already been evaluated 

1520 if value._cascading: 

1521 util.warn( 

1522 "Use of @declared_attr.cascading only applies to " 

1523 "Declarative 'mixin' and 'abstract' classes. " 

1524 "Currently, this flag is ignored on mapped class " 

1525 "%s" % self.cls 

1526 ) 

1527 

1528 value = getattr(cls, k) 

1529 

1530 elif ( 

1531 isinstance(value, QueryableAttribute) 

1532 and value.class_ is not cls 

1533 and value.key != k 

1534 ): 

1535 # detect a QueryableAttribute that's already mapped being 

1536 # assigned elsewhere in userland, turn into a synonym() 

1537 value = SynonymProperty(value.key) 

1538 setattr(cls, k, value) 

1539 

1540 if k in ("metadata", "registry"): 

1541 noun = { 

1542 "metadata": "MetaData instance", 

1543 "registry": "Declarative class registry", 

1544 } 

1545 util.warn( 

1546 f"Attribute name '{k}' should be left reserved for the " 

1547 f"{noun[k]} when using a Declarative class." 

1548 ) 

1549 

1550 if ( 

1551 isinstance(value, tuple) 

1552 and len(value) == 1 

1553 and isinstance(value[0], (Column, _MappedAttribute)) 

1554 ): 

1555 util.warn( 

1556 "Ignoring declarative-like tuple value of attribute " 

1557 "'%s': possibly a copy-and-paste error with a comma " 

1558 "accidentally placed at the end of the line?" % k 

1559 ) 

1560 continue 

1561 elif look_for_dataclass_things and isinstance( 

1562 value, dataclasses.Field 

1563 ): 

1564 # we collected a dataclass Field; dataclasses would have 

1565 # set up the correct state on the class 

1566 continue 

1567 elif not isinstance(value, (Column, _DCAttributeOptions)): 

1568 # using @declared_attr for some object that 

1569 # isn't Column/MapperProperty/_DCAttributeOptions; remove 

1570 # from the clsdict_view 

1571 # and place the evaluated value onto the class. 

1572 collected_attributes.pop(k) 

1573 self._warn_for_decl_attributes(cls, k, value) 

1574 if not late_mapped: 

1575 setattr(cls, k, value) 

1576 continue 

1577 elif isinstance(value, Column): 

1578 _undefer_column_name( 

1579 k, self.column_copies.get(value, value) # type: ignore[arg-type] # noqa: E501 

1580 ) 

1581 else: 

1582 if isinstance(value, _IntrospectsAnnotations): 

1583 ( 

1584 annotation, 

1585 mapped_container, 

1586 extracted_mapped_annotation, 

1587 is_dataclass, 

1588 attr_value, 

1589 originating_module, 

1590 originating_class, 

1591 ) = self.collected_annotations.get( 

1592 k, (None, None, None, False, None, None, None) 

1593 ) 

1594 

1595 # issue #8692 - don't do any annotation interpretation if 

1596 # an annotation were present and a container such as 

1597 # Mapped[] etc. were not used. If annotation is None, 

1598 # do declarative_scan so that the property can raise 

1599 # for required 

1600 if ( 

1601 mapped_container is not None 

1602 or annotation is None 

1603 # issue #10516: need to do declarative_scan even with 

1604 # a non-Mapped annotation if we are doing 

1605 # __allow_unmapped__, for things like col.name 

1606 # assignment 

1607 or allow_unmapped_annotations 

1608 ): 

1609 try: 

1610 value.declarative_scan( 

1611 self, 

1612 self.registry, 

1613 cls, 

1614 originating_module, 

1615 k, 

1616 mapped_container, 

1617 annotation, 

1618 extracted_mapped_annotation, 

1619 is_dataclass, 

1620 ) 

1621 except NameError as ne: 

1622 raise orm_exc.MappedAnnotationError( 

1623 f"Could not resolve all types within mapped " 

1624 f'annotation: "{annotation}". Ensure all ' 

1625 f"types are written correctly and are " 

1626 f"imported within the module in use." 

1627 ) from ne 

1628 else: 

1629 # assert that we were expecting annotations 

1630 # without Mapped[] were going to be passed. 

1631 # otherwise an error should have been raised 

1632 # by util._extract_mapped_subtype before we got here. 

1633 assert expect_annotations_wo_mapped 

1634 

1635 if isinstance(value, _DCAttributeOptions): 

1636 if ( 

1637 value._has_dataclass_arguments 

1638 and not look_for_dataclass_things 

1639 ): 

1640 if isinstance(value, MapperProperty): 

1641 argnames = [ 

1642 "init", 

1643 "default_factory", 

1644 "repr", 

1645 "default", 

1646 "dataclass_metadata", 

1647 ] 

1648 else: 

1649 argnames = [ 

1650 "init", 

1651 "default_factory", 

1652 "repr", 

1653 "dataclass_metadata", 

1654 ] 

1655 

1656 args = { 

1657 a 

1658 for a in argnames 

1659 if getattr( 

1660 value._attribute_options, f"dataclasses_{a}" 

1661 ) 

1662 is not _NoArg.NO_ARG 

1663 } 

1664 

1665 raise exc.ArgumentError( 

1666 f"Attribute '{k}' on class {cls} includes " 

1667 f"dataclasses argument(s): " 

1668 f"{', '.join(sorted(repr(a) for a in args))} but " 

1669 f"class does not specify " 

1670 "SQLAlchemy native dataclass configuration." 

1671 ) 

1672 

1673 if not isinstance(value, (MapperProperty, _MapsColumns)): 

1674 # filter for _DCAttributeOptions objects that aren't 

1675 # MapperProperty / mapped_column(). Currently this 

1676 # includes AssociationProxy. pop it from the things 

1677 # we're going to map and set it up as a descriptor 

1678 # on the class. 

1679 collected_attributes.pop(k) 

1680 

1681 # Assoc Prox (or other descriptor object that may 

1682 # use _DCAttributeOptions) is usually here, except if 

1683 # 1. we're a 

1684 # dataclass, dataclasses would have removed the 

1685 # attr here or 2. assoc proxy is coming from a 

1686 # superclass, we want it to be direct here so it 

1687 # tracks state or 3. assoc prox comes from 

1688 # declared_attr, uncommon case 

1689 setattr(cls, k, value) 

1690 continue 

1691 

1692 our_stuff[k] = value 

1693 

1694 def _extract_declared_columns(self) -> None: 

1695 our_stuff = self.properties 

1696 

1697 # extract columns from the class dict 

1698 declared_columns = self.declared_columns 

1699 column_ordering = self.column_ordering 

1700 name_to_prop_key = collections.defaultdict(set) 

1701 

1702 for key, c in list(our_stuff.items()): 

1703 if isinstance(c, _MapsColumns): 

1704 mp_to_assign = c.mapper_property_to_assign 

1705 if mp_to_assign: 

1706 our_stuff[key] = mp_to_assign 

1707 else: 

1708 # if no mapper property to assign, this currently means 

1709 # this is a MappedColumn that will produce a Column for us 

1710 del our_stuff[key] 

1711 

1712 for col, sort_order in c.columns_to_assign: 

1713 if not isinstance(c, CompositeProperty): 

1714 name_to_prop_key[col.name].add(key) 

1715 declared_columns.add(col) 

1716 

1717 # we would assert this, however we want the below 

1718 # warning to take effect instead. See #9630 

1719 # assert col not in column_ordering 

1720 

1721 column_ordering[col] = sort_order 

1722 

1723 # if this is a MappedColumn and the attribute key we 

1724 # have is not what the column has for its key, map the 

1725 # Column explicitly under the attribute key name. 

1726 # otherwise, Mapper will map it under the column key. 

1727 if mp_to_assign is None and key != col.key: 

1728 our_stuff[key] = col 

1729 elif isinstance(c, Column): 

1730 # undefer previously occurred here, and now occurs earlier. 

1731 # ensure every column we get here has been named 

1732 assert c.name is not None 

1733 name_to_prop_key[c.name].add(key) 

1734 declared_columns.add(c) 

1735 # if the column is the same name as the key, 

1736 # remove it from the explicit properties dict. 

1737 # the normal rules for assigning column-based properties 

1738 # will take over, including precedence of columns 

1739 # in multi-column ColumnProperties. 

1740 if key == c.key: 

1741 del our_stuff[key] 

1742 

1743 for name, keys in name_to_prop_key.items(): 

1744 if len(keys) > 1: 

1745 util.warn( 

1746 "On class %r, Column object %r named " 

1747 "directly multiple times, " 

1748 "only one will be used: %s. " 

1749 "Consider using orm.synonym instead" 

1750 % (self.classname, name, (", ".join(sorted(keys)))) 

1751 ) 

1752 

1753 def _setup_table(self, table: Optional[FromClause] = None) -> None: 

1754 cls = self.cls 

1755 cls_as_Decl = cast("MappedClassProtocol[Any]", cls) 

1756 

1757 tablename = self.tablename 

1758 table_args = self.table_args 

1759 clsdict_view = self.clsdict_view 

1760 declared_columns = self.declared_columns 

1761 column_ordering = self.column_ordering 

1762 

1763 manager = attributes.manager_of_class(cls) 

1764 

1765 if ( 

1766 self.table_fn is None 

1767 and "__table__" not in clsdict_view 

1768 and table is None 

1769 ): 

1770 if hasattr(cls, "__table_cls__"): 

1771 table_cls = cast( 

1772 Type[Table], 

1773 util.unbound_method_to_callable(cls.__table_cls__), # type: ignore[no-untyped-call] # noqa: E501 

1774 ) 

1775 else: 

1776 table_cls = Table 

1777 

1778 if tablename is not None: 

1779 args: Tuple[Any, ...] = () 

1780 table_kw: Dict[str, Any] = {} 

1781 

1782 if table_args: 

1783 if isinstance(table_args, dict): 

1784 table_kw = table_args 

1785 elif isinstance(table_args, tuple): 

1786 if isinstance(table_args[-1], dict): 

1787 args, table_kw = table_args[0:-1], table_args[-1] 

1788 else: 

1789 args = table_args 

1790 

1791 autoload_with = clsdict_view.get("__autoload_with__") 

1792 if autoload_with: 

1793 table_kw["autoload_with"] = autoload_with 

1794 

1795 autoload = clsdict_view.get("__autoload__") 

1796 if autoload: 

1797 table_kw["autoload"] = True 

1798 

1799 sorted_columns = sorted( 

1800 declared_columns, 

1801 key=lambda c: column_ordering.get(c, 0), 

1802 ) 

1803 table = self.set_cls_attribute( 

1804 "__table__", 

1805 table_cls( 

1806 tablename, 

1807 self._metadata_for_cls(manager), 

1808 *sorted_columns, 

1809 *args, 

1810 **table_kw, 

1811 ), 

1812 ) 

1813 else: 

1814 if table is None: 

1815 if self.table_fn: 

1816 table = self.set_cls_attribute( 

1817 "__table__", self.table_fn() 

1818 ) 

1819 else: 

1820 table = cls_as_Decl.__table__ 

1821 if declared_columns: 

1822 for c in declared_columns: 

1823 if not table.c.contains_column(c): 

1824 raise exc.ArgumentError( 

1825 "Can't add additional column %r when " 

1826 "specifying __table__" % c.key 

1827 ) 

1828 

1829 self.local_table = table 

1830 

1831 def _metadata_for_cls(self, manager: ClassManager[Any]) -> MetaData: 

1832 return _metadata_for_cls_fn(self.cls, manager.registry) 

1833 

1834 def _setup_inheriting_mapper(self) -> None: 

1835 cls = self.cls 

1836 

1837 inherits = None 

1838 

1839 if inherits is None: 

1840 # since we search for classical mappings now, search for 

1841 # multiple mapped bases as well and raise an error. 

1842 inherits_search = [] 

1843 for base_ in cls.__bases__: 

1844 c = _resolve_for_abstract_or_classical(base_) 

1845 if c is None: 

1846 continue 

1847 

1848 if _is_supercls_for_inherits(c) and c not in inherits_search: 

1849 inherits_search.append(c) 

1850 

1851 if inherits_search: 

1852 if len(inherits_search) > 1: 

1853 raise exc.InvalidRequestError( 

1854 "Class %s has multiple mapped bases: %r" 

1855 % (cls, inherits_search) 

1856 ) 

1857 inherits = inherits_search[0] 

1858 elif isinstance(inherits, Mapper): 

1859 inherits = inherits.class_ 

1860 

1861 self.inherits = inherits 

1862 

1863 clsdict_view = self.clsdict_view 

1864 if "__table__" not in clsdict_view and self.tablename is None: 

1865 self.single = True 

1866 

1867 def _setup_inheriting_columns(self) -> None: 

1868 table = self.local_table 

1869 cls = self.cls 

1870 table_args = self.table_args 

1871 declared_columns = self.declared_columns 

1872 

1873 if ( 

1874 table is None 

1875 and self.inherits is None 

1876 and not _get_immediate_cls_attr(cls, "__no_table__") 

1877 ): 

1878 raise exc.InvalidRequestError( 

1879 "Class %r does not have a __table__ or __tablename__ " 

1880 "specified and does not inherit from an existing " 

1881 "table-mapped class." % cls 

1882 ) 

1883 elif self.inherits: 

1884 inherited_mapper_or_config = _declared_mapping_info(self.inherits) 

1885 assert inherited_mapper_or_config is not None 

1886 inherited_table = inherited_mapper_or_config.local_table 

1887 inherited_persist_selectable = ( 

1888 inherited_mapper_or_config.persist_selectable 

1889 ) 

1890 

1891 if table is None: 

1892 # single table inheritance. 

1893 # ensure no table args 

1894 if table_args: 

1895 raise exc.ArgumentError( 

1896 "Can't place __table_args__ on an inherited class " 

1897 "with no table." 

1898 ) 

1899 

1900 # add any columns declared here to the inherited table. 

1901 if declared_columns and not isinstance(inherited_table, Table): 

1902 raise exc.ArgumentError( 

1903 f"Can't declare columns on single-table-inherited " 

1904 f"subclass {self.cls}; superclass {self.inherits} " 

1905 "is not mapped to a Table" 

1906 ) 

1907 

1908 for col in declared_columns: 

1909 assert inherited_table is not None 

1910 if col.name in inherited_table.c: 

1911 if inherited_table.c[col.name] is col: 

1912 continue 

1913 raise exc.ArgumentError( 

1914 f"Column '{col}' on class {cls.__name__} " 

1915 f"conflicts with existing column " 

1916 f"'{inherited_table.c[col.name]}'. If using " 

1917 f"Declarative, consider using the " 

1918 "use_existing_column parameter of mapped_column() " 

1919 "to resolve conflicts." 

1920 ) 

1921 if col.primary_key: 

1922 raise exc.ArgumentError( 

1923 "Can't place primary key columns on an inherited " 

1924 "class with no table." 

1925 ) 

1926 

1927 if TYPE_CHECKING: 

1928 assert isinstance(inherited_table, Table) 

1929 

1930 inherited_table.append_column(col) 

1931 if ( 

1932 inherited_persist_selectable is not None 

1933 and inherited_persist_selectable is not inherited_table 

1934 ): 

1935 inherited_persist_selectable._refresh_for_new_column( 

1936 col 

1937 ) 

1938 

1939 def _prepare_mapper_arguments(self, mapper_kw: _MapperKwArgs) -> None: 

1940 properties = self.properties 

1941 

1942 if self.mapper_args_fn: 

1943 mapper_args = self.mapper_args_fn() 

1944 else: 

1945 mapper_args = {} 

1946 

1947 if mapper_kw: 

1948 mapper_args.update(mapper_kw) 

1949 

1950 if "properties" in mapper_args: 

1951 properties = dict(properties) 

1952 properties.update(mapper_args["properties"]) 

1953 

1954 # make sure that column copies are used rather 

1955 # than the original columns from any mixins 

1956 for k in ("version_id_col", "polymorphic_on"): 

1957 if k in mapper_args: 

1958 v = mapper_args[k] 

1959 mapper_args[k] = self.column_copies.get(v, v) 

1960 

1961 if "primary_key" in mapper_args: 

1962 mapper_args["primary_key"] = [ 

1963 self.column_copies.get(v, v) 

1964 for v in util.to_list(mapper_args["primary_key"]) 

1965 ] 

1966 

1967 if "inherits" in mapper_args: 

1968 inherits_arg = mapper_args["inherits"] 

1969 if isinstance(inherits_arg, Mapper): 

1970 inherits_arg = inherits_arg.class_ 

1971 

1972 if inherits_arg is not self.inherits: 

1973 raise exc.InvalidRequestError( 

1974 "mapper inherits argument given for non-inheriting " 

1975 "class %s" % (mapper_args["inherits"]) 

1976 ) 

1977 

1978 if self.inherits: 

1979 mapper_args["inherits"] = self.inherits 

1980 

1981 if self.inherits and not mapper_args.get("concrete", False): 

1982 # note the superclass is expected to have a Mapper assigned and 

1983 # not be a deferred config, as this is called within map() 

1984 inherited_mapper = class_mapper(self.inherits, False) 

1985 inherited_table = inherited_mapper.local_table 

1986 

1987 # single or joined inheritance 

1988 # exclude any cols on the inherited table which are 

1989 # not mapped on the parent class, to avoid 

1990 # mapping columns specific to sibling/nephew classes 

1991 if "exclude_properties" not in mapper_args: 

1992 mapper_args["exclude_properties"] = exclude_properties = { 

1993 c.key 

1994 for c in inherited_table.c 

1995 if c not in inherited_mapper._columntoproperty 

1996 }.union(inherited_mapper.exclude_properties or ()) 

1997 exclude_properties.difference_update( 

1998 [c.key for c in self.declared_columns] 

1999 ) 

2000 

2001 # look through columns in the current mapper that 

2002 # are keyed to a propname different than the colname 

2003 # (if names were the same, we'd have popped it out above, 

2004 # in which case the mapper makes this combination). 

2005 # See if the superclass has a similar column property. 

2006 # If so, join them together. 

2007 for k, col in list(properties.items()): 

2008 if not isinstance(col, expression.ColumnElement): 

2009 continue 

2010 if k in inherited_mapper._props: 

2011 p = inherited_mapper._props[k] 

2012 if isinstance(p, ColumnProperty): 

2013 # note here we place the subclass column 

2014 # first. See [ticket:1892] for background. 

2015 properties[k] = [col] + p.columns 

2016 result_mapper_args = mapper_args.copy() 

2017 result_mapper_args["properties"] = properties 

2018 self.mapper_args = result_mapper_args 

2019 

2020 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]: 

2021 self._prepare_mapper_arguments(mapper_kw) 

2022 if hasattr(self.cls, "__mapper_cls__"): 

2023 mapper_cls = cast( 

2024 "Type[Mapper[Any]]", 

2025 util.unbound_method_to_callable( 

2026 self.cls.__mapper_cls__ # type: ignore[no-untyped-call] 

2027 ), 

2028 ) 

2029 else: 

2030 mapper_cls = Mapper 

2031 

2032 return self.set_cls_attribute( 

2033 "__mapper__", 

2034 mapper_cls(self.cls, self.local_table, **self.mapper_args), 

2035 ) 

2036 

2037 

2038class _UnmappedDataclassConfig(_ClassScanAbstractConfig): 

2039 """Configurator that will produce an unmapped dataclass.""" 

2040 

2041 __slots__ = ( 

2042 "clsdict_view", 

2043 "collected_attributes", 

2044 "collected_annotations", 

2045 "allow_dataclass_fields", 

2046 "dataclass_setup_arguments", 

2047 "is_dataclass_prior_to_mapping", 

2048 "allow_unmapped_annotations", 

2049 ) 

2050 

2051 def __init__( 

2052 self, 

2053 cls_: Type[_O], 

2054 dict_: _ClassDict, 

2055 ): 

2056 super().__init__(cls_) 

2057 self.clsdict_view = ( 

2058 util.immutabledict(dict_) if dict_ else util.EMPTY_DICT 

2059 ) 

2060 self.dataclass_setup_arguments = getattr( 

2061 self.cls, "_sa_apply_dc_transforms", None 

2062 ) 

2063 

2064 self.is_dataclass_prior_to_mapping = dataclasses.is_dataclass(cls_) 

2065 self.allow_dataclass_fields = False 

2066 self.allow_unmapped_annotations = True 

2067 self.collected_attributes = {} 

2068 self.collected_annotations = {} 

2069 

2070 self._scan_attributes() 

2071 

2072 self._setup_dataclasses_transforms( 

2073 enable_descriptor_defaults=False, revert=True 

2074 ) 

2075 

2076 def _scan_attributes(self) -> None: 

2077 cls = self.cls 

2078 

2079 clsdict_view = self.clsdict_view 

2080 collected_attributes = self.collected_attributes 

2081 _include_dunders = self._include_dunders 

2082 

2083 attribute_is_overridden = self._cls_attr_override_checker(self.cls) 

2084 

2085 local_attributes_for_class = self._cls_attr_resolver(cls) 

2086 for ( 

2087 name, 

2088 obj, 

2089 annotation, 

2090 is_dataclass_field, 

2091 ) in local_attributes_for_class(): 

2092 if name in _include_dunders: 

2093 continue 

2094 elif is_dataclass_field and ( 

2095 name not in clsdict_view or clsdict_view[name] is not obj 

2096 ): 

2097 # here, we are definitely looking at the target class 

2098 # and not a superclass. this is currently a 

2099 # dataclass-only path. if the name is only 

2100 # a dataclass field and isn't in local cls.__dict__, 

2101 # put the object there. 

2102 # assert that the dataclass-enabled resolver agrees 

2103 # with what we are seeing 

2104 

2105 assert not attribute_is_overridden(name, obj) 

2106 

2107 if _is_declarative_props(obj): 

2108 obj = obj.fget() 

2109 

2110 collected_attributes[name] = obj 

2111 self._collect_annotation(name, annotation, cls, False, obj) 

2112 else: 

2113 self._collect_annotation(name, annotation, cls, None, obj) 

2114 if name in clsdict_view: 

2115 collected_attributes[name] = obj 

2116 

2117 

2118@util.preload_module("sqlalchemy.orm.decl_api") 

2119def _as_dc_declaredattr( 

2120 field_metadata: Mapping[str, Any], sa_dataclass_metadata_key: str 

2121) -> Any: 

2122 # wrap lambdas inside dataclass fields inside an ad-hoc declared_attr. 

2123 # we can't write it because field.metadata is immutable :( so we have 

2124 # to go through extra trouble to compare these 

2125 decl_api = util.preloaded.orm_decl_api 

2126 obj = field_metadata[sa_dataclass_metadata_key] 

2127 if callable(obj) and not isinstance(obj, decl_api.declared_attr): 

2128 return decl_api.declared_attr(obj) 

2129 else: 

2130 return obj 

2131 

2132 

2133class _DeferredDeclarativeConfig(_DeclarativeMapperConfig): 

2134 """Configurator that extends _DeclarativeMapperConfig to add a 

2135 "deferred" step, to allow extensions like AbstractConcreteBase, 

2136 DeferredMapping to partially set up a mapping that is "prepared" 

2137 when table metadata is ready. 

2138 

2139 """ 

2140 

2141 _cls: weakref.ref[Type[Any]] 

2142 

2143 is_deferred = True 

2144 

2145 _configs: util.OrderedDict[ 

2146 weakref.ref[Type[Any]], _DeferredDeclarativeConfig 

2147 ] = util.OrderedDict() 

2148 

2149 def _early_mapping(self, mapper_kw: _MapperKwArgs) -> None: 

2150 pass 

2151 

2152 @property 

2153 def cls(self) -> Type[Any]: 

2154 return self._cls() # type: ignore[return-value] 

2155 

2156 @cls.setter 

2157 def cls(self, class_: Type[Any]) -> None: 

2158 self._cls = weakref.ref(class_, self._remove_config_cls) 

2159 self._configs[self._cls] = self 

2160 

2161 @classmethod 

2162 def _remove_config_cls(cls, ref: weakref.ref[Type[Any]]) -> None: 

2163 cls._configs.pop(ref, None) 

2164 

2165 @classmethod 

2166 def has_cls(cls, class_: Type[Any]) -> bool: 

2167 # 2.6 fails on weakref if class_ is an old style class 

2168 return isinstance(class_, type) and weakref.ref(class_) in cls._configs 

2169 

2170 @classmethod 

2171 def raise_unmapped_for_cls(cls, class_: Type[Any]) -> NoReturn: 

2172 if hasattr(class_, "_sa_raise_deferred_config"): 

2173 class_._sa_raise_deferred_config() 

2174 

2175 raise orm_exc.UnmappedClassError( 

2176 class_, 

2177 msg=( 

2178 f"Class {orm_exc._safe_cls_name(class_)} has a deferred " 

2179 "mapping on it. It is not yet usable as a mapped class." 

2180 ), 

2181 ) 

2182 

2183 @classmethod 

2184 def config_for_cls(cls, class_: Type[Any]) -> _DeferredDeclarativeConfig: 

2185 return cls._configs[weakref.ref(class_)] 

2186 

2187 @classmethod 

2188 def classes_for_base( 

2189 cls, base_cls: Type[Any], sort: bool = True 

2190 ) -> List[_DeferredDeclarativeConfig]: 

2191 classes_for_base = [ 

2192 m 

2193 for m, cls_ in [(m, m.cls) for m in cls._configs.values()] 

2194 if cls_ is not None and issubclass(cls_, base_cls) 

2195 ] 

2196 

2197 if not sort: 

2198 return classes_for_base 

2199 

2200 all_m_by_cls = {m.cls: m for m in classes_for_base} 

2201 

2202 tuples: List[ 

2203 Tuple[_DeferredDeclarativeConfig, _DeferredDeclarativeConfig] 

2204 ] = [] 

2205 for m_cls in all_m_by_cls: 

2206 tuples.extend( 

2207 (all_m_by_cls[base_cls], all_m_by_cls[m_cls]) 

2208 for base_cls in m_cls.__bases__ 

2209 if base_cls in all_m_by_cls 

2210 ) 

2211 return list(topological.sort(tuples, classes_for_base)) 

2212 

2213 def map(self, mapper_kw: _MapperKwArgs = util.EMPTY_DICT) -> Mapper[Any]: 

2214 self._configs.pop(self._cls, None) 

2215 return super().map(mapper_kw) 

2216 

2217 

2218def _add_attribute( 

2219 cls: Type[Any], key: str, value: MapperProperty[Any] 

2220) -> None: 

2221 """add an attribute to an existing declarative class. 

2222 

2223 This runs through the logic to determine MapperProperty, 

2224 adds it to the Mapper, adds a column to the mapped Table, etc. 

2225 

2226 """ 

2227 

2228 if "__mapper__" in cls.__dict__: 

2229 mapped_cls = cast("MappedClassProtocol[Any]", cls) 

2230 

2231 def _table_or_raise(mc: MappedClassProtocol[Any]) -> Table: 

2232 if isinstance(mc.__table__, Table): 

2233 return mc.__table__ 

2234 raise exc.InvalidRequestError( 

2235 f"Cannot add a new attribute to mapped class {mc.__name__!r} " 

2236 "because it's not mapped against a table." 

2237 ) 

2238 

2239 if isinstance(value, Column): 

2240 _undefer_column_name(key, value) 

2241 _table_or_raise(mapped_cls).append_column( 

2242 value, replace_existing=True 

2243 ) 

2244 mapped_cls.__mapper__.add_property(key, value) 

2245 elif isinstance(value, _MapsColumns): 

2246 mp = value.mapper_property_to_assign 

2247 for col, _ in value.columns_to_assign: 

2248 _undefer_column_name(key, col) 

2249 _table_or_raise(mapped_cls).append_column( 

2250 col, replace_existing=True 

2251 ) 

2252 if not mp: 

2253 mapped_cls.__mapper__.add_property(key, col) 

2254 if mp: 

2255 mapped_cls.__mapper__.add_property(key, mp) 

2256 elif isinstance(value, MapperProperty): 

2257 mapped_cls.__mapper__.add_property(key, value) 

2258 elif isinstance(value, QueryableAttribute) and value.key != key: 

2259 # detect a QueryableAttribute that's already mapped being 

2260 # assigned elsewhere in userland, turn into a synonym() 

2261 value = SynonymProperty(value.key) 

2262 mapped_cls.__mapper__.add_property(key, value) 

2263 else: 

2264 type.__setattr__(cls, key, value) 

2265 mapped_cls.__mapper__._expire_memoizations() 

2266 else: 

2267 type.__setattr__(cls, key, value) 

2268 

2269 

2270def _del_attribute(cls: Type[Any], key: str) -> None: 

2271 if ( 

2272 "__mapper__" in cls.__dict__ 

2273 and key in cls.__dict__ 

2274 and not cast( 

2275 "MappedClassProtocol[Any]", cls 

2276 ).__mapper__._dispose_called 

2277 ): 

2278 value = cls.__dict__[key] 

2279 if isinstance( 

2280 value, (Column, _MapsColumns, MapperProperty, QueryableAttribute) 

2281 ): 

2282 raise NotImplementedError( 

2283 "Can't un-map individual mapped attributes on a mapped class." 

2284 ) 

2285 else: 

2286 type.__delattr__(cls, key) 

2287 cast( 

2288 "MappedClassProtocol[Any]", cls 

2289 ).__mapper__._expire_memoizations() 

2290 else: 

2291 type.__delattr__(cls, key) 

2292 

2293 

2294def _declarative_constructor(self: Any, **kwargs: Any) -> None: 

2295 """A simple constructor that allows initialization from kwargs. 

2296 

2297 Sets attributes on the constructed instance using the names and 

2298 values in ``kwargs``. 

2299 

2300 Only keys that are present as 

2301 attributes of the instance's class are allowed. These could be, 

2302 for example, any mapped columns or relationships. 

2303 """ 

2304 cls_ = type(self) 

2305 for k in kwargs: 

2306 if not hasattr(cls_, k): 

2307 raise TypeError( 

2308 "%r is an invalid keyword argument for %s" % (k, cls_.__name__) 

2309 ) 

2310 setattr(self, k, kwargs[k]) 

2311 

2312 

2313_declarative_constructor.__name__ = "__init__" 

2314 

2315 

2316def _undefer_column_name(key: str, column: Column[Any]) -> None: 

2317 if column.key is None: 

2318 column.key = key 

2319 if column.name is None: 

2320 column.name = key