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

1# orm/properties.py 

2# Copyright (C) 2005-2025 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 List 

21from typing import Optional 

22from typing import Sequence 

23from typing import Set 

24from typing import Tuple 

25from typing import Type 

26from typing import TYPE_CHECKING 

27from typing import TypeVar 

28from typing import Union 

29 

30from . import attributes 

31from . import exc as orm_exc 

32from . import strategy_options 

33from .base import _DeclarativeMapped 

34from .base import class_mapper 

35from .descriptor_props import CompositeProperty 

36from .descriptor_props import ConcreteInheritedProperty 

37from .descriptor_props import SynonymProperty 

38from .interfaces import _AttributeOptions 

39from .interfaces import _DataclassDefaultsDontSet 

40from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS 

41from .interfaces import _IntrospectsAnnotations 

42from .interfaces import _MapsColumns 

43from .interfaces import MapperProperty 

44from .interfaces import PropComparator 

45from .interfaces import StrategizedProperty 

46from .relationships import RelationshipProperty 

47from .util import de_stringify_annotation 

48from .. import exc as sa_exc 

49from .. import ForeignKey 

50from .. import log 

51from .. import util 

52from ..sql import coercions 

53from ..sql import roles 

54from ..sql.base import _NoArg 

55from ..sql.schema import Column 

56from ..sql.schema import SchemaConst 

57from ..sql.type_api import TypeEngine 

58from ..util.typing import de_optionalize_union_types 

59from ..util.typing import get_args 

60from ..util.typing import includes_none 

61from ..util.typing import is_a_type 

62from ..util.typing import is_fwd_ref 

63from ..util.typing import is_pep593 

64from ..util.typing import is_pep695 

65from ..util.typing import Self 

66 

67if TYPE_CHECKING: 

68 from ._typing import _IdentityKeyType 

69 from ._typing import _InstanceDict 

70 from ._typing import _ORMColumnExprArgument 

71 from ._typing import _RegistryType 

72 from .base import Mapped 

73 from .decl_base import _ClassScanMapperConfig 

74 from .mapper import Mapper 

75 from .session import Session 

76 from .state import _InstallLoaderCallableProto 

77 from .state import InstanceState 

78 from ..sql._typing import _InfoType 

79 from ..sql.elements import ColumnElement 

80 from ..sql.elements import NamedColumn 

81 from ..sql.operators import OperatorType 

82 from ..util.typing import _AnnotationScanType 

83 from ..util.typing import RODescriptorReference 

84 

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

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

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

88 

89__all__ = [ 

90 "ColumnProperty", 

91 "CompositeProperty", 

92 "ConcreteInheritedProperty", 

93 "RelationshipProperty", 

94 "SynonymProperty", 

95] 

96 

97 

98@log.class_logger 

99class ColumnProperty( 

100 _DataclassDefaultsDontSet, 

101 _MapsColumns[_T], 

102 StrategizedProperty[_T], 

103 _IntrospectsAnnotations, 

104 log.Identified, 

105): 

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

107 or other column expression. 

108 

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

110 

111 """ 

112 

113 strategy_wildcard_key = strategy_options._COLUMN_TOKEN 

114 inherit_cache = True 

115 """:meta private:""" 

116 

117 _links_to_entity = False 

118 

119 columns: List[NamedColumn[Any]] 

120 

121 _is_polymorphic_discriminator: bool 

122 

123 _mapped_by_synonym: Optional[str] 

124 

125 comparator_factory: Type[PropComparator[_T]] 

126 

127 __slots__ = ( 

128 "columns", 

129 "group", 

130 "deferred", 

131 "instrument", 

132 "comparator_factory", 

133 "active_history", 

134 "expire_on_flush", 

135 "_default_scalar_value", 

136 "_creation_order", 

137 "_is_polymorphic_discriminator", 

138 "_mapped_by_synonym", 

139 "_deferred_column_loader", 

140 "_raise_column_loader", 

141 "_renders_in_subqueries", 

142 "raiseload", 

143 ) 

144 

145 def __init__( 

146 self, 

147 column: _ORMColumnExprArgument[_T], 

148 *additional_columns: _ORMColumnExprArgument[Any], 

149 attribute_options: Optional[_AttributeOptions] = None, 

150 group: Optional[str] = None, 

151 deferred: bool = False, 

152 raiseload: bool = False, 

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

154 active_history: bool = False, 

155 default_scalar_value: Any = None, 

156 expire_on_flush: bool = True, 

157 info: Optional[_InfoType] = None, 

158 doc: Optional[str] = None, 

159 _instrument: bool = True, 

160 _assume_readonly_dc_attributes: bool = False, 

161 ): 

162 super().__init__( 

163 attribute_options=attribute_options, 

164 _assume_readonly_dc_attributes=_assume_readonly_dc_attributes, 

165 ) 

166 columns = (column,) + additional_columns 

167 self.columns = [ 

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

169 ] 

170 self.group = group 

171 self.deferred = deferred 

172 self.raiseload = raiseload 

173 self.instrument = _instrument 

174 self.comparator_factory = ( 

175 comparator_factory 

176 if comparator_factory is not None 

177 else self.__class__.Comparator 

178 ) 

179 self.active_history = active_history 

180 self._default_scalar_value = default_scalar_value 

181 self.expire_on_flush = expire_on_flush 

182 

183 if info is not None: 

184 self.info.update(info) 

185 

186 if doc is not None: 

187 self.doc = doc 

188 else: 

189 for col in reversed(self.columns): 

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

191 if doc is not None: 

192 self.doc = doc 

193 break 

194 else: 

195 self.doc = None 

196 

197 util.set_creation_order(self) 

198 

199 self.strategy_key = ( 

200 ("deferred", self.deferred), 

201 ("instrument", self.instrument), 

202 ) 

203 if self.raiseload: 

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

205 

206 def declarative_scan( 

207 self, 

208 decl_scan: _ClassScanMapperConfig, 

209 registry: _RegistryType, 

210 cls: Type[Any], 

211 originating_module: Optional[str], 

212 key: str, 

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

214 annotation: Optional[_AnnotationScanType], 

215 extracted_mapped_annotation: Optional[_AnnotationScanType], 

216 is_dataclass_field: bool, 

217 ) -> None: 

218 column = self.columns[0] 

219 if column.key is None: 

220 column.key = key 

221 if column.name is None: 

222 column.name = key 

223 

224 @property 

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

226 return self 

227 

228 @property 

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

230 # mypy doesn't care about the isinstance here 

231 return [ 

232 (c, 0) # type: ignore 

233 for c in self.columns 

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

235 ] 

236 

237 def _memoized_attr__renders_in_subqueries(self) -> bool: 

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

239 return self.strategy._have_default_expression # type: ignore 

240 

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

242 self not in self.parent._readonly_props 

243 ) 

244 

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

246 def _memoized_attr__deferred_column_loader( 

247 self, 

248 ) -> _InstallLoaderCallableProto[Any]: 

249 state = util.preloaded.orm_state 

250 strategies = util.preloaded.orm_strategies 

251 return state.InstanceState._instance_level_callable_processor( 

252 self.parent.class_manager, 

253 strategies._LoadDeferredColumns(self.key), 

254 self.key, 

255 ) 

256 

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

258 def _memoized_attr__raise_column_loader( 

259 self, 

260 ) -> _InstallLoaderCallableProto[Any]: 

261 state = util.preloaded.orm_state 

262 strategies = util.preloaded.orm_strategies 

263 return state.InstanceState._instance_level_callable_processor( 

264 self.parent.class_manager, 

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

266 self.key, 

267 ) 

268 

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

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

271 into an instrumented attribute. 

272 """ 

273 

274 return self.expression 

275 

276 @property 

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

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

279 

280 E.g.:: 

281 

282 

283 class File(Base): 

284 # ... 

285 

286 name = Column(String(64)) 

287 extension = Column(String(8)) 

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

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

290 

291 .. seealso:: 

292 

293 :ref:`mapper_column_property_sql_expressions_composed` 

294 

295 """ 

296 return self.columns[0] 

297 

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

299 if not self.instrument: 

300 return 

301 

302 attributes._register_descriptor( 

303 mapper.class_, 

304 self.key, 

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

306 parententity=mapper, 

307 doc=self.doc, 

308 ) 

309 

310 def do_init(self) -> None: 

311 super().do_init() 

312 

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

314 self.columns 

315 ): 

316 util.warn( 

317 ( 

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

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

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

321 "mapped attribute name." 

322 ) 

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

324 ) 

325 

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

327 return ColumnProperty( 

328 *self.columns, 

329 deferred=self.deferred, 

330 group=self.group, 

331 active_history=self.active_history, 

332 default_scalar_value=self._default_scalar_value, 

333 ) 

334 

335 def merge( 

336 self, 

337 session: Session, 

338 source_state: InstanceState[Any], 

339 source_dict: _InstanceDict, 

340 dest_state: InstanceState[Any], 

341 dest_dict: _InstanceDict, 

342 load: bool, 

343 _recursive: Dict[Any, object], 

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

345 ) -> None: 

346 if not self.instrument: 

347 return 

348 elif self.key in source_dict: 

349 value = source_dict[self.key] 

350 

351 if not load: 

352 dest_dict[self.key] = value 

353 else: 

354 impl = dest_state.get_impl(self.key) 

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

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

357 dest_state._expire_attributes( 

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

359 ) 

360 

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

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

363 :class:`.ColumnProperty` attributes. 

364 

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

366 overview. 

367 

368 .. seealso:: 

369 

370 :class:`.PropComparator` 

371 

372 :class:`.ColumnOperators` 

373 

374 :ref:`types_operators` 

375 

376 :attr:`.TypeEngine.comparator_factory` 

377 

378 """ 

379 

380 if not TYPE_CHECKING: 

381 # prevent pylance from being clever about slots 

382 __slots__ = "__clause_element__", "info", "expressions" 

383 

384 prop: RODescriptorReference[ColumnProperty[_PT]] 

385 

386 expressions: Sequence[NamedColumn[Any]] 

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

388 attribute, adjusted for any aliasing in progress. 

389 

390 .. seealso:: 

391 

392 :ref:`maptojoin` - usage example 

393 """ 

394 

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

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

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

398 

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

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

401 annotations to give hints to various ORM functions as to 

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

403 to the mapper's with_polymorphic selectable if one is 

404 present. 

405 

406 """ 

407 

408 pe = self._parententity 

409 annotations: Dict[str, Any] = { 

410 "entity_namespace": pe, 

411 "parententity": pe, 

412 "parentmapper": pe, 

413 "proxy_key": self.prop.key, 

414 } 

415 

416 col = column 

417 

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

419 # the column against the polymorphic selectable. 

420 # see also orm.util._orm_downgrade_polymorphic_columns 

421 # for the reverse operation. 

422 if self._parentmapper._polymorphic_adapter: 

423 mapper_local_col = col 

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

425 

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

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

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

429 annotations["adapt_column"] = mapper_local_col 

430 

431 return col._annotate(annotations)._set_propagate_attrs( 

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

433 ) 

434 

435 if TYPE_CHECKING: 

436 

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

438 

439 def _memoized_method___clause_element__( 

440 self, 

441 ) -> NamedColumn[_PT]: 

442 if self.adapter: 

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

444 else: 

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

446 

447 def _memoized_attr_info(self) -> _InfoType: 

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

449 

450 ce = self.__clause_element__() 

451 try: 

452 return ce.info # type: ignore 

453 except AttributeError: 

454 return self.prop.info 

455 

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

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

458 attribute, adjusted for any aliasing in progress. 

459 

460 """ 

461 if self.adapter: 

462 return [ 

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

464 for col in self.prop.columns 

465 ] 

466 else: 

467 return [ 

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

469 ] 

470 

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

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

473 

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

475 """ 

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

477 

478 def operate( 

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

480 ) -> ColumnElement[Any]: 

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

482 

483 def reverse_operate( 

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

485 ) -> ColumnElement[Any]: 

486 col = self.__clause_element__() 

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

488 

489 def __str__(self) -> str: 

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

491 return object.__repr__(self) 

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

493 

494 

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

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

497 

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

499 

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

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

502 

503 .. seealso:: 

504 

505 :class:`.MappedColumn` 

506 

507 """ 

508 

509 inherit_cache = True 

510 """:meta private:""" 

511 

512 

513class MappedColumn( 

514 _DataclassDefaultsDontSet, 

515 _IntrospectsAnnotations, 

516 _MapsColumns[_T], 

517 _DeclarativeMapped[_T], 

518): 

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

520 

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

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

523 configuration. 

524 

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

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

527 

528 .. versionadded:: 2.0 

529 

530 

531 """ 

532 

533 __slots__ = ( 

534 "column", 

535 "_creation_order", 

536 "_sort_order", 

537 "foreign_keys", 

538 "_has_nullable", 

539 "_has_insert_default", 

540 "deferred", 

541 "deferred_group", 

542 "deferred_raiseload", 

543 "active_history", 

544 "_default_scalar_value", 

545 "_attribute_options", 

546 "_has_dataclass_arguments", 

547 "_use_existing_column", 

548 ) 

549 

550 deferred: Union[_NoArg, bool] 

551 deferred_raiseload: bool 

552 deferred_group: Optional[str] 

553 

554 column: Column[_T] 

555 foreign_keys: Optional[Set[ForeignKey]] 

556 _attribute_options: _AttributeOptions 

557 

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

559 self._attribute_options = attr_opts = kw.pop( 

560 "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS 

561 ) 

562 

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

564 

565 self._has_dataclass_arguments = ( 

566 attr_opts is not None 

567 and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS 

568 and any( 

569 attr_opts[i] is not _NoArg.NO_ARG 

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

571 if attr != "dataclasses_default" 

572 ) 

573 ) 

574 

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

576 self._has_insert_default = insert_default is not _NoArg.NO_ARG 

577 self._default_scalar_value = _NoArg.NO_ARG 

578 

579 if attr_opts.dataclasses_default is not _NoArg.NO_ARG: 

580 kw["default"] = attr_opts.dataclasses_default 

581 

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

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

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

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

586 

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

588 

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

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

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

592 # dataclass role 

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

594 

595 self.foreign_keys = self.column.foreign_keys 

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

597 None, 

598 SchemaConst.NULL_UNSPECIFIED, 

599 ) 

600 util.set_creation_order(self) 

601 

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

603 new = self.__class__.__new__(self.__class__) 

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

605 new.deferred = self.deferred 

606 new.deferred_group = self.deferred_group 

607 new.deferred_raiseload = self.deferred_raiseload 

608 new.foreign_keys = new.column.foreign_keys 

609 new.active_history = self.active_history 

610 new._has_nullable = self._has_nullable 

611 new._attribute_options = self._attribute_options 

612 new._has_insert_default = self._has_insert_default 

613 new._has_dataclass_arguments = self._has_dataclass_arguments 

614 new._use_existing_column = self._use_existing_column 

615 new._sort_order = self._sort_order 

616 new._default_scalar_value = self._default_scalar_value 

617 util.set_creation_order(new) 

618 return new 

619 

620 @property 

621 def name(self) -> str: 

622 return self.column.name 

623 

624 @property 

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

626 effective_deferred = self.deferred 

627 if effective_deferred is _NoArg.NO_ARG: 

628 effective_deferred = bool( 

629 self.deferred_group or self.deferred_raiseload 

630 ) 

631 

632 if ( 

633 effective_deferred 

634 or self.active_history 

635 or self._default_scalar_value is not _NoArg.NO_ARG 

636 ): 

637 return ColumnProperty( 

638 self.column, 

639 deferred=effective_deferred, 

640 group=self.deferred_group, 

641 raiseload=self.deferred_raiseload, 

642 attribute_options=self._attribute_options, 

643 active_history=self.active_history, 

644 default_scalar_value=( 

645 self._default_scalar_value 

646 if self._default_scalar_value is not _NoArg.NO_ARG 

647 else None 

648 ), 

649 ) 

650 else: 

651 return None 

652 

653 @property 

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

655 return [ 

656 ( 

657 self.column, 

658 ( 

659 self._sort_order 

660 if self._sort_order is not _NoArg.NO_ARG 

661 else 0 

662 ), 

663 ) 

664 ] 

665 

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

667 return self.column 

668 

669 def operate( 

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

671 ) -> ColumnElement[Any]: 

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

673 

674 def reverse_operate( 

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

676 ) -> ColumnElement[Any]: 

677 col = self.__clause_element__() 

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

679 

680 def found_in_pep593_annotated(self) -> Any: 

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

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

683 return MappedColumn() 

684 

685 def declarative_scan( 

686 self, 

687 decl_scan: _ClassScanMapperConfig, 

688 registry: _RegistryType, 

689 cls: Type[Any], 

690 originating_module: Optional[str], 

691 key: str, 

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

693 annotation: Optional[_AnnotationScanType], 

694 extracted_mapped_annotation: Optional[_AnnotationScanType], 

695 is_dataclass_field: bool, 

696 ) -> None: 

697 column = self.column 

698 

699 if ( 

700 self._use_existing_column 

701 and decl_scan.inherits 

702 and decl_scan.single 

703 ): 

704 if decl_scan.is_deferred: 

705 raise sa_exc.ArgumentError( 

706 "Can't use use_existing_column with deferred mappers" 

707 ) 

708 supercls_mapper = class_mapper(decl_scan.inherits, False) 

709 

710 colname = column.name if column.name is not None else key 

711 column = self.column = supercls_mapper.local_table.c.get( # type: ignore[assignment] # noqa: E501 

712 colname, column 

713 ) 

714 

715 if column.key is None: 

716 column.key = key 

717 if column.name is None: 

718 column.name = key 

719 

720 sqltype = column.type 

721 

722 if extracted_mapped_annotation is None: 

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

724 self._raise_for_required(key, cls) 

725 else: 

726 return 

727 

728 self._init_column_for_annotation( 

729 cls, 

730 registry, 

731 extracted_mapped_annotation, 

732 originating_module, 

733 ) 

734 

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

736 def declarative_scan_for_composite( 

737 self, 

738 registry: _RegistryType, 

739 cls: Type[Any], 

740 originating_module: Optional[str], 

741 key: str, 

742 param_name: str, 

743 param_annotation: _AnnotationScanType, 

744 ) -> None: 

745 decl_base = util.preloaded.orm_decl_base 

746 decl_base._undefer_column_name(param_name, self.column) 

747 self._init_column_for_annotation( 

748 cls, registry, param_annotation, originating_module 

749 ) 

750 

751 def _init_column_for_annotation( 

752 self, 

753 cls: Type[Any], 

754 registry: _RegistryType, 

755 argument: _AnnotationScanType, 

756 originating_module: Optional[str], 

757 ) -> None: 

758 sqltype = self.column.type 

759 

760 if is_fwd_ref( 

761 argument, check_generic=True, check_for_plain_string=True 

762 ): 

763 assert originating_module is not None 

764 argument = de_stringify_annotation( 

765 cls, argument, originating_module, include_generic=True 

766 ) 

767 

768 nullable = includes_none(argument) 

769 

770 if not self._has_nullable: 

771 self.column.nullable = nullable 

772 

773 our_type = de_optionalize_union_types(argument) 

774 

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

776 our_type_is_pep593 = False 

777 raw_pep_593_type = None 

778 

779 if is_pep593(our_type): 

780 our_type_is_pep593 = True 

781 

782 pep_593_components = get_args(our_type) 

783 raw_pep_593_type = pep_593_components[0] 

784 if nullable: 

785 raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) 

786 find_mapped_in = pep_593_components[1:] 

787 elif is_pep695(argument) and is_pep593(argument.__value__): 

788 # do not support nested annotation inside unions ets 

789 find_mapped_in = get_args(argument.__value__)[1:] 

790 

791 use_args_from: Optional[MappedColumn[Any]] 

792 for elem in find_mapped_in: 

793 if isinstance(elem, MappedColumn): 

794 use_args_from = elem 

795 break 

796 else: 

797 use_args_from = None 

798 

799 if use_args_from is not None: 

800 

801 if ( 

802 self._has_insert_default 

803 or self._attribute_options.dataclasses_default 

804 is not _NoArg.NO_ARG 

805 ): 

806 omit_defaults = True 

807 else: 

808 omit_defaults = False 

809 

810 use_args_from.column._merge( 

811 self.column, omit_defaults=omit_defaults 

812 ) 

813 sqltype = self.column.type 

814 

815 if ( 

816 use_args_from.deferred is not _NoArg.NO_ARG 

817 and self.deferred is _NoArg.NO_ARG 

818 ): 

819 self.deferred = use_args_from.deferred 

820 

821 if ( 

822 use_args_from.deferred_group is not None 

823 and self.deferred_group is None 

824 ): 

825 self.deferred_group = use_args_from.deferred_group 

826 

827 if ( 

828 use_args_from.deferred_raiseload is not None 

829 and self.deferred_raiseload is None 

830 ): 

831 self.deferred_raiseload = use_args_from.deferred_raiseload 

832 

833 if ( 

834 use_args_from._use_existing_column 

835 and not self._use_existing_column 

836 ): 

837 self._use_existing_column = True 

838 

839 if use_args_from.active_history: 

840 self.active_history = use_args_from.active_history 

841 

842 if ( 

843 use_args_from._sort_order is not None 

844 and self._sort_order is _NoArg.NO_ARG 

845 ): 

846 self._sort_order = use_args_from._sort_order 

847 

848 if ( 

849 use_args_from.column.key is not None 

850 or use_args_from.column.name is not None 

851 ): 

852 util.warn_deprecated( 

853 "Can't use the 'key' or 'name' arguments in " 

854 "Annotated with mapped_column(); this will be ignored", 

855 "2.0.22", 

856 ) 

857 

858 if use_args_from._has_dataclass_arguments: 

859 for idx, arg in enumerate( 

860 use_args_from._attribute_options._fields 

861 ): 

862 if ( 

863 use_args_from._attribute_options[idx] 

864 is not _NoArg.NO_ARG 

865 ): 

866 arg = arg.replace("dataclasses_", "") 

867 util.warn_deprecated( 

868 f"Argument '{arg}' is a dataclass argument and " 

869 "cannot be specified within a mapped_column() " 

870 "bundled inside of an Annotated object", 

871 "2.0.22", 

872 ) 

873 

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