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

1# orm/properties.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"""MapperProperty implementations. 

9 

10This is a private module which defines the behavior of individual ORM- 

11mapped attributes. 

12 

13""" 

14 

15from __future__ import annotations 

16 

17from typing import Any 

18from typing import cast 

19from typing import Dict 

20from typing import get_args 

21from typing import List 

22from typing import Optional 

23from typing import Sequence 

24from typing import Set 

25from typing import Tuple 

26from typing import Type 

27from typing import TYPE_CHECKING 

28from typing import TypeVar 

29from typing import Union 

30 

31from . import attributes 

32from . import exc as orm_exc 

33from .base import _DeclarativeMapped 

34from .base import class_mapper 

35from .base import Unmapped 

36from .descriptor_props import CompositeProperty 

37from .descriptor_props import ConcreteInheritedProperty 

38from .descriptor_props import SynonymProperty 

39from .interfaces import _AttributeOptions 

40from .interfaces import _DataclassDefaultsDontSet 

41from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS 

42from .interfaces import _IntrospectsAnnotations 

43from .interfaces import _MapsColumns 

44from .interfaces import MapperProperty 

45from .interfaces import PropComparator 

46from .interfaces import StrategizedProperty 

47from .path_registry import _COLUMN_TOKEN 

48from .relationships import RelationshipProperty 

49from .util import de_stringify_annotation 

50from .. import exc as sa_exc 

51from .. import ForeignKey 

52from .. import log 

53from .. import util 

54from ..sql import coercions 

55from ..sql import roles 

56from ..sql.base import _NoArg 

57from ..sql.schema import Column 

58from ..sql.schema import SchemaConst 

59from ..sql.type_api import TypeEngine 

60from ..util.typing import de_optionalize_union_types 

61from ..util.typing import includes_none 

62from ..util.typing import is_a_type 

63from ..util.typing import is_fwd_ref 

64from ..util.typing import is_pep593 

65from ..util.typing import is_pep695 

66from ..util.typing import Self 

67 

68if TYPE_CHECKING: 

69 from typing import ForwardRef 

70 

71 from ._typing import _IdentityKeyType 

72 from ._typing import _InstanceDict 

73 from ._typing import _ORMColumnExprArgument 

74 from ._typing import _RegistryType 

75 from .base import Mapped 

76 from .decl_base import _DeclarativeMapperConfig 

77 from .mapper import Mapper 

78 from .session import Session 

79 from .state import _InstallLoaderCallableProto 

80 from .state import InstanceState 

81 from ..sql._typing import _InfoType 

82 from ..sql.elements import ColumnElement 

83 from ..sql.elements import NamedColumn 

84 from ..sql.operators import OperatorType 

85 from ..util.typing import _AnnotationScanType 

86 from ..util.typing import _MatchedOnType 

87 from ..util.typing import RODescriptorReference 

88 

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

90_PT = TypeVar("_PT", bound=Any) 

91_NC = TypeVar("_NC", bound="NamedColumn[Any]") 

92_MC = TypeVar("_MC", bound="MappedColumn[Any]") 

93 

94__all__ = [ 

95 "ColumnProperty", 

96 "CompositeProperty", 

97 "ConcreteInheritedProperty", 

98 "RelationshipProperty", 

99 "SynonymProperty", 

100] 

101 

102 

103@log.class_logger 

104class ColumnProperty( 

105 _DataclassDefaultsDontSet, 

106 _MapsColumns[_T], 

107 StrategizedProperty[_T], 

108 _IntrospectsAnnotations, 

109 log.Identified, 

110): 

111 """Describes an object attribute that corresponds to a table column 

112 or other column expression. 

113 

114 Public constructor is the :func:`_orm.column_property` function. 

115 

116 """ 

117 

118 strategy_wildcard_key = _COLUMN_TOKEN 

119 inherit_cache = True 

120 """:meta private:""" 

121 

122 _links_to_entity = False 

123 

124 columns: List[NamedColumn[Any]] 

125 

126 _is_polymorphic_discriminator: bool 

127 

128 _mapped_by_synonym: Optional[str] 

129 

130 comparator_factory: Type[PropComparator[_T]] 

131 

132 __slots__ = ( 

133 "columns", 

134 "group", 

135 "deferred", 

136 "instrument", 

137 "comparator_factory", 

138 "active_history", 

139 "expire_on_flush", 

140 "_default_scalar_value", 

141 "_creation_order", 

142 "_is_polymorphic_discriminator", 

143 "_mapped_by_synonym", 

144 "_deferred_column_loader", 

145 "_raise_column_loader", 

146 "_renders_in_subqueries", 

147 "raiseload", 

148 ) 

149 

150 def __init__( 

151 self, 

152 column: _ORMColumnExprArgument[_T], 

153 *additional_columns: _ORMColumnExprArgument[Any], 

154 attribute_options: Optional[_AttributeOptions] = None, 

155 group: Optional[str] = None, 

156 deferred: bool = False, 

157 raiseload: bool = False, 

158 comparator_factory: Optional[Type[PropComparator[_T]]] = None, 

159 active_history: bool = False, 

160 default_scalar_value: Any = None, 

161 expire_on_flush: bool = True, 

162 info: Optional[_InfoType] = None, 

163 doc: Optional[str] = None, 

164 _instrument: bool = True, 

165 _assume_readonly_dc_attributes: bool = False, 

166 ): 

167 super().__init__( 

168 attribute_options=attribute_options, 

169 _assume_readonly_dc_attributes=_assume_readonly_dc_attributes, 

170 ) 

171 columns = (column,) + additional_columns 

172 self.columns = [ 

173 coercions.expect(roles.LabeledColumnExprRole, c) for c in columns 

174 ] 

175 self.group = group 

176 self.deferred = deferred 

177 self.raiseload = raiseload 

178 self.instrument = _instrument 

179 self.comparator_factory = ( 

180 comparator_factory 

181 if comparator_factory is not None 

182 else self.__class__.Comparator 

183 ) 

184 self.active_history = active_history 

185 self._default_scalar_value = default_scalar_value 

186 self.expire_on_flush = expire_on_flush 

187 

188 if info is not None: 

189 self.info.update(info) 

190 

191 if doc is not None: 

192 self.doc = doc 

193 else: 

194 for col in reversed(self.columns): 

195 doc = getattr(col, "doc", None) 

196 if doc is not None: 

197 self.doc = doc 

198 break 

199 else: 

200 self.doc = None 

201 

202 util.set_creation_order(self) 

203 

204 self.strategy_key = ( 

205 ("deferred", self.deferred), 

206 ("instrument", self.instrument), 

207 ) 

208 if self.raiseload: 

209 self.strategy_key += (("raiseload", True),) 

210 

211 def declarative_scan( 

212 self, 

213 decl_scan: _DeclarativeMapperConfig, 

214 registry: _RegistryType, 

215 cls: Type[Any], 

216 originating_module: Optional[str], 

217 key: str, 

218 mapped_container: Optional[Type[Mapped[Any]]], 

219 annotation: Optional[_AnnotationScanType], 

220 extracted_mapped_annotation: Optional[_AnnotationScanType], 

221 is_dataclass_field: bool, 

222 ) -> None: 

223 column = self.columns[0] 

224 if column.key is None: 

225 column.key = key 

226 if column.name is None: 

227 column.name = key 

228 

229 @property 

230 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]: 

231 return self 

232 

233 @property 

234 def columns_to_assign(self) -> List[Tuple[Column[Any], int]]: 

235 # mypy doesn't care about the isinstance here 

236 return [ 

237 (c, 0) # type: ignore[misc] 

238 for c in self.columns 

239 if isinstance(c, Column) and c.table is None 

240 ] 

241 

242 def _memoized_attr__renders_in_subqueries(self) -> bool: 

243 if ("query_expression", True) in self.strategy_key: 

244 return self.strategy._have_default_expression # type: ignore[attr-defined, no-any-return] # noqa: E501 

245 

246 return ("deferred", True) not in self.strategy_key or ( 

247 self not in self.parent._readonly_props 

248 ) 

249 

250 @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies") 

251 def _memoized_attr__deferred_column_loader( 

252 self, 

253 ) -> _InstallLoaderCallableProto[Any]: 

254 state = util.preloaded.orm_state 

255 strategies = util.preloaded.orm_strategies 

256 return state.InstanceState._instance_level_callable_processor( 

257 self.parent.class_manager, 

258 strategies._LoadDeferredColumns(self.key), 

259 self.key, 

260 ) 

261 

262 @util.preload_module("sqlalchemy.orm.state", "sqlalchemy.orm.strategies") 

263 def _memoized_attr__raise_column_loader( 

264 self, 

265 ) -> _InstallLoaderCallableProto[Any]: 

266 state = util.preloaded.orm_state 

267 strategies = util.preloaded.orm_strategies 

268 return state.InstanceState._instance_level_callable_processor( 

269 self.parent.class_manager, 

270 strategies._LoadDeferredColumns(self.key, True), 

271 self.key, 

272 ) 

273 

274 def __clause_element__(self) -> roles.ColumnsClauseRole: 

275 """Allow the ColumnProperty to work in expression before it is turned 

276 into an instrumented attribute. 

277 """ 

278 

279 return self.expression 

280 

281 @property 

282 def expression(self) -> roles.ColumnsClauseRole: 

283 """Return the primary column or expression for this ColumnProperty. 

284 

285 E.g.:: 

286 

287 

288 class File(Base): 

289 # ... 

290 

291 name = Column(String(64)) 

292 extension = Column(String(8)) 

293 filename = column_property(name + "." + extension) 

294 path = column_property("C:/" + filename.expression) 

295 

296 .. seealso:: 

297 

298 :ref:`mapper_column_property_sql_expressions_composed` 

299 

300 """ 

301 return self.columns[0] 

302 

303 def instrument_class(self, mapper: Mapper[Any]) -> None: 

304 if not self.instrument: 

305 return 

306 

307 attributes._register_descriptor( 

308 mapper.class_, 

309 self.key, 

310 comparator=self.comparator_factory(self, mapper), 

311 parententity=mapper, 

312 doc=self.doc, 

313 ) 

314 

315 def do_init(self) -> None: 

316 super().do_init() 

317 

318 if len(self.columns) > 1 and set(self.parent.primary_key).issuperset( 

319 self.columns 

320 ): 

321 util.warn( 

322 ( 

323 "On mapper %s, primary key column '%s' is being combined " 

324 "with distinct primary key column '%s' in attribute '%s'. " 

325 "Use explicit properties to give each column its own " 

326 "mapped attribute name." 

327 ) 

328 % (self.parent, self.columns[1], self.columns[0], self.key) 

329 ) 

330 

331 def copy(self) -> ColumnProperty[_T]: 

332 return ColumnProperty( 

333 *self.columns, 

334 deferred=self.deferred, 

335 group=self.group, 

336 active_history=self.active_history, 

337 default_scalar_value=self._default_scalar_value, 

338 ) 

339 

340 def merge( 

341 self, 

342 session: Session, 

343 source_state: InstanceState[Any], 

344 source_dict: _InstanceDict, 

345 dest_state: InstanceState[Any], 

346 dest_dict: _InstanceDict, 

347 load: bool, 

348 _recursive: Dict[Any, object], 

349 _resolve_conflict_map: Dict[_IdentityKeyType[Any], object], 

350 ) -> None: 

351 if not self.instrument: 

352 return 

353 elif self.key in source_dict: 

354 value = source_dict[self.key] 

355 

356 if not load: 

357 dest_dict[self.key] = value 

358 else: 

359 impl = dest_state.get_impl(self.key) 

360 impl.set(dest_state, dest_dict, value, None) 

361 elif dest_state.has_identity and self.key not in dest_dict: 

362 dest_state._expire_attributes( 

363 dest_dict, [self.key], no_loader=True 

364 ) 

365 

366 class Comparator(util.MemoizedSlots, PropComparator[_PT]): 

367 """Produce boolean, comparison, and other operators for 

368 :class:`.ColumnProperty` attributes. 

369 

370 See the documentation for :class:`.PropComparator` for a brief 

371 overview. 

372 

373 .. seealso:: 

374 

375 :class:`.PropComparator` 

376 

377 :class:`.ColumnOperators` 

378 

379 :ref:`types_operators` 

380 

381 :attr:`.TypeEngine.comparator_factory` 

382 

383 """ 

384 

385 if not TYPE_CHECKING: 

386 # prevent pylance from being clever about slots 

387 __slots__ = "__clause_element__", "info", "expressions" 

388 

389 prop: RODescriptorReference[ColumnProperty[_PT]] 

390 

391 expressions: Sequence[NamedColumn[Any]] 

392 """The full sequence of columns referenced by this 

393 attribute, adjusted for any aliasing in progress. 

394 

395 .. seealso:: 

396 

397 :ref:`maptojoin` - usage example 

398 """ 

399 

400 def _orm_annotate_column(self, column: _NC) -> _NC: 

401 """annotate and possibly adapt a column to be returned 

402 as the mapped-attribute exposed version of the column. 

403 

404 The column in this context needs to act as much like the 

405 column in an ORM mapped context as possible, so includes 

406 annotations to give hints to various ORM functions as to 

407 the source entity of this column. It also adapts it 

408 to the mapper's with_polymorphic selectable if one is 

409 present. 

410 

411 """ 

412 

413 pe = self._parententity 

414 annotations: Dict[str, Any] = { 

415 "entity_namespace": pe, 

416 "parententity": pe, 

417 "parentmapper": pe, 

418 "proxy_key": self.prop.key, 

419 } 

420 

421 col = column 

422 

423 # for a mapper with polymorphic_on and an adapter, return 

424 # the column against the polymorphic selectable. 

425 # see also orm.util._orm_downgrade_polymorphic_columns 

426 # for the reverse operation. 

427 if self._parentmapper._polymorphic_adapter: 

428 mapper_local_col = col 

429 col = self._parentmapper._polymorphic_adapter.traverse(col) 

430 

431 # this is a clue to the ORM Query etc. that this column 

432 # was adapted to the mapper's polymorphic_adapter. the 

433 # ORM uses this hint to know which column its adapting. 

434 annotations["adapt_column"] = mapper_local_col 

435 

436 return col._annotate(annotations)._set_propagate_attrs( 

437 {"compile_state_plugin": "orm", "plugin_subject": pe} 

438 ) 

439 

440 if TYPE_CHECKING: 

441 

442 def __clause_element__(self) -> NamedColumn[_PT]: ... 

443 

444 def _memoized_method___clause_element__( 

445 self, 

446 ) -> NamedColumn[_PT]: 

447 if self.adapter: 

448 return self.adapter(self.prop.columns[0], self.prop.key) 

449 else: 

450 return self._orm_annotate_column(self.prop.columns[0]) 

451 

452 def _memoized_attr_info(self) -> _InfoType: 

453 """The .info dictionary for this attribute.""" 

454 

455 ce = self.__clause_element__() 

456 try: 

457 return ce.info # type: ignore[no-any-return] 

458 except AttributeError: 

459 return self.prop.info 

460 

461 def _memoized_attr_expressions(self) -> Sequence[NamedColumn[Any]]: 

462 """The full sequence of columns referenced by this 

463 attribute, adjusted for any aliasing in progress. 

464 

465 """ 

466 if self.adapter: 

467 return [ 

468 self.adapter(col, self.prop.key) 

469 for col in self.prop.columns 

470 ] 

471 else: 

472 return [ 

473 self._orm_annotate_column(col) for col in self.prop.columns 

474 ] 

475 

476 def _fallback_getattr(self, key: str) -> Any: 

477 """proxy attribute access down to the mapped column. 

478 

479 this allows user-defined comparison methods to be accessed. 

480 """ 

481 return getattr(self.__clause_element__(), key) 

482 

483 def operate( 

484 self, op: OperatorType, *other: Any, **kwargs: Any 

485 ) -> ColumnElement[Any]: 

486 return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501 

487 

488 def reverse_operate( 

489 self, op: OperatorType, other: Any, **kwargs: Any 

490 ) -> ColumnElement[Any]: 

491 col = self.__clause_element__() 

492 return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501 

493 

494 def __str__(self) -> str: 

495 if not self.parent or not self.key: 

496 return object.__repr__(self) 

497 return str(self.parent.class_.__name__) + "." + self.key 

498 

499 

500class MappedSQLExpression(ColumnProperty[_T], _DeclarativeMapped[_T]): 

501 """Declarative front-end for the :class:`.ColumnProperty` class. 

502 

503 Public constructor is the :func:`_orm.column_property` function. 

504 

505 .. versionchanged:: 2.0 Added :class:`_orm.MappedSQLExpression` as 

506 a Declarative compatible subclass for :class:`_orm.ColumnProperty`. 

507 

508 .. seealso:: 

509 

510 :class:`.MappedColumn` 

511 

512 """ 

513 

514 inherit_cache = True 

515 """:meta private:""" 

516 

517 

518class MappedColumn( 

519 _DataclassDefaultsDontSet, 

520 _IntrospectsAnnotations, 

521 _MapsColumns[_T], 

522 _DeclarativeMapped[_T], 

523): 

524 """Maps a single :class:`_schema.Column` on a class. 

525 

526 :class:`_orm.MappedColumn` is a specialization of the 

527 :class:`_orm.ColumnProperty` class and is oriented towards declarative 

528 configuration. 

529 

530 To construct :class:`_orm.MappedColumn` objects, use the 

531 :func:`_orm.mapped_column` constructor function. 

532 

533 .. versionadded:: 2.0 

534 

535 

536 """ 

537 

538 __slots__ = ( 

539 "column", 

540 "_creation_order", 

541 "_sort_order", 

542 "foreign_keys", 

543 "_has_nullable", 

544 "_has_insert_default", 

545 "deferred", 

546 "deferred_group", 

547 "deferred_raiseload", 

548 "active_history", 

549 "_default_scalar_value", 

550 "_attribute_options", 

551 "_has_dataclass_arguments", 

552 "_use_existing_column", 

553 ) 

554 

555 deferred: Union[_NoArg, bool] 

556 deferred_raiseload: bool 

557 deferred_group: Optional[str] 

558 

559 column: Column[_T] 

560 foreign_keys: Optional[Set[ForeignKey]] 

561 _attribute_options: _AttributeOptions 

562 

563 def __init__(self, *arg: Any, **kw: Any): 

564 self._attribute_options = attr_opts = kw.pop( 

565 "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS 

566 ) 

567 

568 self._use_existing_column = kw.pop("use_existing_column", False) 

569 

570 self._has_dataclass_arguments = ( 

571 attr_opts is not None 

572 and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS 

573 and any( 

574 attr_opts[i] is not _NoArg.NO_ARG 

575 for i, attr in enumerate(attr_opts._fields) 

576 if attr != "dataclasses_default" 

577 ) 

578 ) 

579 

580 insert_default = kw.get("insert_default", _NoArg.NO_ARG) 

581 self._has_insert_default = insert_default is not _NoArg.NO_ARG 

582 self._default_scalar_value = _NoArg.NO_ARG 

583 

584 if attr_opts.dataclasses_default is not _NoArg.NO_ARG: 

585 kw["default"] = attr_opts.dataclasses_default 

586 

587 self.deferred_group = kw.pop("deferred_group", None) 

588 self.deferred_raiseload = kw.pop("deferred_raiseload", None) 

589 self.deferred = kw.pop("deferred", _NoArg.NO_ARG) 

590 self.active_history = kw.pop("active_history", False) 

591 

592 self._sort_order = kw.pop("sort_order", _NoArg.NO_ARG) 

593 

594 # note that this populates "default" into the Column, so that if 

595 # we are a dataclass and "default" is a dataclass default, it is still 

596 # used as a Core-level default for the Column in addition to its 

597 # dataclass role 

598 self.column = cast("Column[_T]", Column(*arg, **kw)) 

599 

600 self.foreign_keys = self.column.foreign_keys 

601 self._has_nullable = "nullable" in kw and kw.get("nullable") not in ( 

602 None, 

603 SchemaConst.NULL_UNSPECIFIED, 

604 ) 

605 util.set_creation_order(self) 

606 

607 def _copy(self, **kw: Any) -> Self: 

608 return self._copy_as(self.__class__, **kw) 

609 

610 def _as_unmapped_column(self) -> UnmappedColumn[_T]: 

611 """produce a copy of this :class:`.MappedColumn` as an 

612 :class:`.UnmappedColumn`, e.g. for a ``mapped_column()`` that's 

613 inside of ``Annotated`` that's applied to ``Unmapped[]``.""" 

614 

615 return self._copy_as(UnmappedColumn) 

616 

617 def _copy_as(self, cls: Type[_MC], **kw: Any) -> _MC: 

618 new = cls.__new__(cls) 

619 new.column = self.column._copy(**kw) 

620 new.deferred = self.deferred 

621 new.deferred_group = self.deferred_group 

622 new.deferred_raiseload = self.deferred_raiseload 

623 new.foreign_keys = new.column.foreign_keys 

624 new.active_history = self.active_history 

625 new._has_nullable = self._has_nullable 

626 new._attribute_options = self._attribute_options 

627 new._has_insert_default = self._has_insert_default 

628 new._has_dataclass_arguments = self._has_dataclass_arguments 

629 new._use_existing_column = self._use_existing_column 

630 new._sort_order = self._sort_order 

631 new._default_scalar_value = self._default_scalar_value 

632 util.set_creation_order(new) 

633 return new 

634 

635 @property 

636 def name(self) -> str: 

637 return self.column.name 

638 

639 @property 

640 def mapper_property_to_assign(self) -> Optional[MapperProperty[_T]]: 

641 effective_deferred = self.deferred 

642 if effective_deferred is _NoArg.NO_ARG: 

643 effective_deferred = bool( 

644 self.deferred_group or self.deferred_raiseload 

645 ) 

646 

647 if ( 

648 effective_deferred 

649 or self.active_history 

650 or self._default_scalar_value is not _NoArg.NO_ARG 

651 ): 

652 return ColumnProperty( 

653 self.column, 

654 deferred=effective_deferred, 

655 group=self.deferred_group, 

656 raiseload=self.deferred_raiseload, 

657 attribute_options=self._attribute_options, 

658 active_history=self.active_history, 

659 default_scalar_value=( 

660 self._default_scalar_value 

661 if self._default_scalar_value is not _NoArg.NO_ARG 

662 else None 

663 ), 

664 ) 

665 else: 

666 return None 

667 

668 @property 

669 def columns_to_assign(self) -> List[Tuple[Column[Any], int]]: 

670 return [ 

671 ( 

672 self.column, 

673 ( 

674 self._sort_order 

675 if self._sort_order is not _NoArg.NO_ARG 

676 else 0 

677 ), 

678 ) 

679 ] 

680 

681 def __clause_element__(self) -> Column[_T]: 

682 return self.column 

683 

684 def operate( 

685 self, op: OperatorType, *other: Any, **kwargs: Any 

686 ) -> ColumnElement[Any]: 

687 return op(self.__clause_element__(), *other, **kwargs) # type: ignore[no-any-return] # noqa: E501 

688 

689 def reverse_operate( 

690 self, op: OperatorType, other: Any, **kwargs: Any 

691 ) -> ColumnElement[Any]: 

692 col = self.__clause_element__() 

693 return op(col._bind_param(op, other), col, **kwargs) # type: ignore[no-any-return] # noqa: E501 

694 

695 def found_in_pep593_annotated(self) -> Any: 

696 # return a blank mapped_column(). This mapped_column()'s 

697 # Column will be merged into it in _init_column_for_annotation(). 

698 return MappedColumn() 

699 

700 def _adjust_for_existing_column( 

701 self, 

702 decl_scan: _DeclarativeMapperConfig, 

703 key: str, 

704 given_column: Column[_T], 

705 ) -> Column[_T]: 

706 if ( 

707 self._use_existing_column 

708 and decl_scan.inherits 

709 and decl_scan.single 

710 ): 

711 if decl_scan.is_deferred: 

712 raise sa_exc.ArgumentError( 

713 "Can't use use_existing_column with deferred mappers" 

714 ) 

715 supercls_mapper = class_mapper(decl_scan.inherits, False) 

716 

717 colname = ( 

718 given_column.name if given_column.name is not None else key 

719 ) 

720 given_column = supercls_mapper.local_table.c.get( # type: ignore[assignment] # noqa: E501 

721 colname, given_column 

722 ) 

723 return given_column 

724 

725 def declarative_scan( 

726 self, 

727 decl_scan: _DeclarativeMapperConfig, 

728 registry: _RegistryType, 

729 cls: Type[Any], 

730 originating_module: Optional[str], 

731 key: str, 

732 mapped_container: Optional[Type[Mapped[Any]]], 

733 annotation: Optional[_AnnotationScanType], 

734 extracted_mapped_annotation: Optional[_AnnotationScanType], 

735 is_dataclass_field: bool, 

736 ) -> None: 

737 column = self.column 

738 

739 column = self.column = self._adjust_for_existing_column( 

740 decl_scan, key, self.column 

741 ) 

742 

743 if column.key is None: 

744 column.key = key 

745 if column.name is None: 

746 column.name = key 

747 

748 sqltype = column.type 

749 

750 if extracted_mapped_annotation is None: 

751 if sqltype._isnull and not self.column.foreign_keys: 

752 self._raise_for_required(key, cls) 

753 else: 

754 return 

755 

756 self._init_column_for_annotation( 

757 cls, 

758 decl_scan, 

759 key, 

760 registry, 

761 extracted_mapped_annotation, 

762 originating_module, 

763 ) 

764 

765 @util.preload_module("sqlalchemy.orm.decl_base") 

766 def declarative_scan_for_composite( 

767 self, 

768 decl_scan: _DeclarativeMapperConfig, 

769 registry: _RegistryType, 

770 cls: Type[Any], 

771 originating_module: Optional[str], 

772 key: str, 

773 param_name: str, 

774 param_annotation: _AnnotationScanType, 

775 ) -> None: 

776 decl_base = util.preloaded.orm_decl_base 

777 decl_base._undefer_column_name(param_name, self.column) 

778 self._init_column_for_annotation( 

779 cls, decl_scan, key, registry, param_annotation, originating_module 

780 ) 

781 

782 def _init_column_for_annotation( 

783 self, 

784 cls: Type[Any], 

785 decl_scan: _DeclarativeMapperConfig, 

786 key: str, 

787 registry: _RegistryType, 

788 argument: _AnnotationScanType, 

789 originating_module: Optional[str], 

790 ) -> None: 

791 sqltype = self.column.type 

792 

793 de_stringified_argument: _MatchedOnType 

794 

795 if is_fwd_ref( 

796 argument, check_generic=True, check_for_plain_string=True 

797 ): 

798 assert originating_module is not None 

799 de_stringified_argument = de_stringify_annotation( 

800 cls, argument, originating_module, include_generic=True 

801 ) 

802 else: 

803 if TYPE_CHECKING: 

804 assert not isinstance(argument, (str, ForwardRef)) 

805 de_stringified_argument = argument 

806 

807 nullable = includes_none(de_stringified_argument) 

808 

809 if not self._has_nullable: 

810 self.column.nullable = nullable 

811 

812 find_mapped_in: Tuple[Any, ...] = () 

813 raw_pep_593_type = resolved_pep_593_type = None 

814 raw_pep_695_type = resolved_pep_695_type = None 

815 

816 our_type: Any = de_optionalize_union_types(de_stringified_argument) 

817 

818 if is_pep695(our_type): 

819 raw_pep_695_type = our_type 

820 our_type = de_optionalize_union_types(raw_pep_695_type.__value__) 

821 our_args = get_args(raw_pep_695_type) 

822 if our_args: 

823 our_type = our_type[our_args] 

824 

825 resolved_pep_695_type = our_type 

826 

827 if is_pep593(our_type): 

828 pep_593_components = get_args(our_type) 

829 raw_pep_593_type = our_type 

830 resolved_pep_593_type = pep_593_components[0] 

831 if nullable: 

832 resolved_pep_593_type = de_optionalize_union_types( 

833 resolved_pep_593_type 

834 ) 

835 find_mapped_in = pep_593_components[1:] 

836 

837 use_args_from: Optional[MappedColumn[Any]] 

838 for elem in find_mapped_in: 

839 if isinstance(elem, MappedColumn): 

840 use_args_from = elem 

841 break 

842 else: 

843 use_args_from = None 

844 

845 if use_args_from is not None: 

846 

847 self.column = use_args_from._adjust_for_existing_column( 

848 decl_scan, key, self.column 

849 ) 

850 

851 if ( 

852 self._has_insert_default 

853 or self._attribute_options.dataclasses_default 

854 is not _NoArg.NO_ARG 

855 ): 

856 omit_defaults = True 

857 else: 

858 omit_defaults = False 

859 

860 use_args_from.column._merge( 

861 self.column, omit_defaults=omit_defaults 

862 ) 

863 sqltype = self.column.type 

864 

865 if ( 

866 use_args_from.deferred is not _NoArg.NO_ARG 

867 and self.deferred is _NoArg.NO_ARG 

868 ): 

869 self.deferred = use_args_from.deferred 

870 

871 if ( 

872 use_args_from.deferred_group is not None 

873 and self.deferred_group is None