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

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 _DEFAULT_ATTRIBUTE_OPTIONS 

40from .interfaces import _IntrospectsAnnotations 

41from .interfaces import _MapsColumns 

42from .interfaces import MapperProperty 

43from .interfaces import PropComparator 

44from .interfaces import StrategizedProperty 

45from .relationships import RelationshipProperty 

46from .util import de_stringify_annotation 

47from .. import exc as sa_exc 

48from .. import ForeignKey 

49from .. import log 

50from .. import util 

51from ..sql import coercions 

52from ..sql import roles 

53from ..sql.base import _NoArg 

54from ..sql.schema import Column 

55from ..sql.schema import SchemaConst 

56from ..sql.type_api import TypeEngine 

57from ..util.typing import de_optionalize_union_types 

58from ..util.typing import get_args 

59from ..util.typing import includes_none 

60from ..util.typing import is_a_type 

61from ..util.typing import is_fwd_ref 

62from ..util.typing import is_pep593 

63from ..util.typing import is_pep695 

64from ..util.typing import Self 

65 

66if TYPE_CHECKING: 

67 from ._typing import _IdentityKeyType 

68 from ._typing import _InstanceDict 

69 from ._typing import _ORMColumnExprArgument 

70 from ._typing import _RegistryType 

71 from .base import Mapped 

72 from .decl_base import _ClassScanMapperConfig 

73 from .mapper import Mapper 

74 from .session import Session 

75 from .state import _InstallLoaderCallableProto 

76 from .state import InstanceState 

77 from ..sql._typing import _InfoType 

78 from ..sql.elements import ColumnElement 

79 from ..sql.elements import NamedColumn 

80 from ..sql.operators import OperatorType 

81 from ..util.typing import _AnnotationScanType 

82 from ..util.typing import RODescriptorReference 

83 

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

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

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

87 

88__all__ = [ 

89 "ColumnProperty", 

90 "CompositeProperty", 

91 "ConcreteInheritedProperty", 

92 "RelationshipProperty", 

93 "SynonymProperty", 

94] 

95 

96 

97@log.class_logger 

98class ColumnProperty( 

99 _MapsColumns[_T], 

100 StrategizedProperty[_T], 

101 _IntrospectsAnnotations, 

102 log.Identified, 

103): 

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

105 or other column expression. 

106 

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

108 

109 """ 

110 

111 strategy_wildcard_key = strategy_options._COLUMN_TOKEN 

112 inherit_cache = True 

113 """:meta private:""" 

114 

115 _links_to_entity = False 

116 

117 columns: List[NamedColumn[Any]] 

118 

119 _is_polymorphic_discriminator: bool 

120 

121 _mapped_by_synonym: Optional[str] 

122 

123 comparator_factory: Type[PropComparator[_T]] 

124 

125 __slots__ = ( 

126 "columns", 

127 "group", 

128 "deferred", 

129 "instrument", 

130 "comparator_factory", 

131 "active_history", 

132 "expire_on_flush", 

133 "_creation_order", 

134 "_is_polymorphic_discriminator", 

135 "_mapped_by_synonym", 

136 "_deferred_column_loader", 

137 "_raise_column_loader", 

138 "_renders_in_subqueries", 

139 "raiseload", 

140 ) 

141 

142 def __init__( 

143 self, 

144 column: _ORMColumnExprArgument[_T], 

145 *additional_columns: _ORMColumnExprArgument[Any], 

146 attribute_options: Optional[_AttributeOptions] = None, 

147 group: Optional[str] = None, 

148 deferred: bool = False, 

149 raiseload: bool = False, 

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

151 active_history: bool = False, 

152 expire_on_flush: bool = True, 

153 info: Optional[_InfoType] = None, 

154 doc: Optional[str] = None, 

155 _instrument: bool = True, 

156 _assume_readonly_dc_attributes: bool = False, 

157 ): 

158 super().__init__( 

159 attribute_options=attribute_options, 

160 _assume_readonly_dc_attributes=_assume_readonly_dc_attributes, 

161 ) 

162 columns = (column,) + additional_columns 

163 self.columns = [ 

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

165 ] 

166 self.group = group 

167 self.deferred = deferred 

168 self.raiseload = raiseload 

169 self.instrument = _instrument 

170 self.comparator_factory = ( 

171 comparator_factory 

172 if comparator_factory is not None 

173 else self.__class__.Comparator 

174 ) 

175 self.active_history = active_history 

176 self.expire_on_flush = expire_on_flush 

177 

178 if info is not None: 

179 self.info.update(info) 

180 

181 if doc is not None: 

182 self.doc = doc 

183 else: 

184 for col in reversed(self.columns): 

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

186 if doc is not None: 

187 self.doc = doc 

188 break 

189 else: 

190 self.doc = None 

191 

192 util.set_creation_order(self) 

193 

194 self.strategy_key = ( 

195 ("deferred", self.deferred), 

196 ("instrument", self.instrument), 

197 ) 

198 if self.raiseload: 

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

200 

201 def declarative_scan( 

202 self, 

203 decl_scan: _ClassScanMapperConfig, 

204 registry: _RegistryType, 

205 cls: Type[Any], 

206 originating_module: Optional[str], 

207 key: str, 

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

209 annotation: Optional[_AnnotationScanType], 

210 extracted_mapped_annotation: Optional[_AnnotationScanType], 

211 is_dataclass_field: bool, 

212 ) -> None: 

213 column = self.columns[0] 

214 if column.key is None: 

215 column.key = key 

216 if column.name is None: 

217 column.name = key 

218 

219 @property 

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

221 return self 

222 

223 @property 

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

225 # mypy doesn't care about the isinstance here 

226 return [ 

227 (c, 0) # type: ignore 

228 for c in self.columns 

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

230 ] 

231 

232 def _memoized_attr__renders_in_subqueries(self) -> bool: 

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

234 return self.strategy._have_default_expression # type: ignore 

235 

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

237 self not in self.parent._readonly_props 

238 ) 

239 

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

241 def _memoized_attr__deferred_column_loader( 

242 self, 

243 ) -> _InstallLoaderCallableProto[Any]: 

244 state = util.preloaded.orm_state 

245 strategies = util.preloaded.orm_strategies 

246 return state.InstanceState._instance_level_callable_processor( 

247 self.parent.class_manager, 

248 strategies.LoadDeferredColumns(self.key), 

249 self.key, 

250 ) 

251 

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

253 def _memoized_attr__raise_column_loader( 

254 self, 

255 ) -> _InstallLoaderCallableProto[Any]: 

256 state = util.preloaded.orm_state 

257 strategies = util.preloaded.orm_strategies 

258 return state.InstanceState._instance_level_callable_processor( 

259 self.parent.class_manager, 

260 strategies.LoadDeferredColumns(self.key, True), 

261 self.key, 

262 ) 

263 

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

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

266 into an instrumented attribute. 

267 """ 

268 

269 return self.expression 

270 

271 @property 

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

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

274 

275 E.g.:: 

276 

277 

278 class File(Base): 

279 # ... 

280 

281 name = Column(String(64)) 

282 extension = Column(String(8)) 

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

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

285 

286 .. seealso:: 

287 

288 :ref:`mapper_column_property_sql_expressions_composed` 

289 

290 """ 

291 return self.columns[0] 

292 

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

294 if not self.instrument: 

295 return 

296 

297 attributes.register_descriptor( 

298 mapper.class_, 

299 self.key, 

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

301 parententity=mapper, 

302 doc=self.doc, 

303 ) 

304 

305 def do_init(self) -> None: 

306 super().do_init() 

307 

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

309 self.columns 

310 ): 

311 util.warn( 

312 ( 

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

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

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

316 "mapped attribute name." 

317 ) 

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

319 ) 

320 

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

322 return ColumnProperty( 

323 *self.columns, 

324 deferred=self.deferred, 

325 group=self.group, 

326 active_history=self.active_history, 

327 ) 

328 

329 def merge( 

330 self, 

331 session: Session, 

332 source_state: InstanceState[Any], 

333 source_dict: _InstanceDict, 

334 dest_state: InstanceState[Any], 

335 dest_dict: _InstanceDict, 

336 load: bool, 

337 _recursive: Dict[Any, object], 

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

339 ) -> None: 

340 if not self.instrument: 

341 return 

342 elif self.key in source_dict: 

343 value = source_dict[self.key] 

344 

345 if not load: 

346 dest_dict[self.key] = value 

347 else: 

348 impl = dest_state.get_impl(self.key) 

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

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

351 dest_state._expire_attributes( 

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

353 ) 

354 

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

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

357 :class:`.ColumnProperty` attributes. 

358 

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

360 overview. 

361 

362 .. seealso:: 

363 

364 :class:`.PropComparator` 

365 

366 :class:`.ColumnOperators` 

367 

368 :ref:`types_operators` 

369 

370 :attr:`.TypeEngine.comparator_factory` 

371 

372 """ 

373 

374 if not TYPE_CHECKING: 

375 # prevent pylance from being clever about slots 

376 __slots__ = "__clause_element__", "info", "expressions" 

377 

378 prop: RODescriptorReference[ColumnProperty[_PT]] 

379 

380 expressions: Sequence[NamedColumn[Any]] 

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

382 attribute, adjusted for any aliasing in progress. 

383 

384 .. versionadded:: 1.3.17 

385 

386 .. seealso:: 

387 

388 :ref:`maptojoin` - usage example 

389 """ 

390 

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

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

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

394 

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

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

397 annotations to give hints to various ORM functions as to 

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

399 to the mapper's with_polymorphic selectable if one is 

400 present. 

401 

402 """ 

403 

404 pe = self._parententity 

405 annotations: Dict[str, Any] = { 

406 "entity_namespace": pe, 

407 "parententity": pe, 

408 "parentmapper": pe, 

409 "proxy_key": self.prop.key, 

410 } 

411 

412 col = column 

413 

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

415 # the column against the polymorphic selectable. 

416 # see also orm.util._orm_downgrade_polymorphic_columns 

417 # for the reverse operation. 

418 if self._parentmapper._polymorphic_adapter: 

419 mapper_local_col = col 

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

421 

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

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

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

425 annotations["adapt_column"] = mapper_local_col 

426 

427 return col._annotate(annotations)._set_propagate_attrs( 

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

429 ) 

430 

431 if TYPE_CHECKING: 

432 

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

434 

435 def _memoized_method___clause_element__( 

436 self, 

437 ) -> NamedColumn[_PT]: 

438 if self.adapter: 

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

440 else: 

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

442 

443 def _memoized_attr_info(self) -> _InfoType: 

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

445 

446 ce = self.__clause_element__() 

447 try: 

448 return ce.info # type: ignore 

449 except AttributeError: 

450 return self.prop.info 

451 

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

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

454 attribute, adjusted for any aliasing in progress. 

455 

456 .. versionadded:: 1.3.17 

457 

458 """ 

459 if self.adapter: 

460 return [ 

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

462 for col in self.prop.columns 

463 ] 

464 else: 

465 return [ 

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

467 ] 

468 

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

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

471 

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

473 """ 

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

475 

476 def operate( 

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

478 ) -> ColumnElement[Any]: 

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

480 

481 def reverse_operate( 

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

483 ) -> ColumnElement[Any]: 

484 col = self.__clause_element__() 

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

486 

487 def __str__(self) -> str: 

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

489 return object.__repr__(self) 

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

491 

492 

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

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

495 

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

497 

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

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

500 

501 .. seealso:: 

502 

503 :class:`.MappedColumn` 

504 

505 """ 

506 

507 inherit_cache = True 

508 """:meta private:""" 

509 

510 

511class MappedColumn( 

512 _IntrospectsAnnotations, 

513 _MapsColumns[_T], 

514 _DeclarativeMapped[_T], 

515): 

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

517 

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

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

520 configuration. 

521 

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

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

524 

525 .. versionadded:: 2.0 

526 

527 

528 """ 

529 

530 __slots__ = ( 

531 "column", 

532 "_creation_order", 

533 "_sort_order", 

534 "foreign_keys", 

535 "_has_nullable", 

536 "_has_insert_default", 

537 "deferred", 

538 "deferred_group", 

539 "deferred_raiseload", 

540 "active_history", 

541 "_attribute_options", 

542 "_has_dataclass_arguments", 

543 "_use_existing_column", 

544 ) 

545 

546 deferred: Union[_NoArg, bool] 

547 deferred_raiseload: bool 

548 deferred_group: Optional[str] 

549 

550 column: Column[_T] 

551 foreign_keys: Optional[Set[ForeignKey]] 

552 _attribute_options: _AttributeOptions 

553 

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

555 self._attribute_options = attr_opts = kw.pop( 

556 "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS 

557 ) 

558 

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

560 

561 self._has_dataclass_arguments = ( 

562 attr_opts is not None 

563 and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS 

564 and any( 

565 attr_opts[i] is not _NoArg.NO_ARG 

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

567 if attr != "dataclasses_default" 

568 ) 

569 ) 

570 

571 insert_default = kw.pop("insert_default", _NoArg.NO_ARG) 

572 self._has_insert_default = insert_default is not _NoArg.NO_ARG 

573 

574 if self._has_insert_default: 

575 kw["default"] = insert_default 

576 elif attr_opts.dataclasses_default is not _NoArg.NO_ARG: 

577 kw["default"] = attr_opts.dataclasses_default 

578 

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

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

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

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

583 

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

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

586 self.foreign_keys = self.column.foreign_keys 

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

588 None, 

589 SchemaConst.NULL_UNSPECIFIED, 

590 ) 

591 util.set_creation_order(self) 

592 

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

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

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

596 new.deferred = self.deferred 

597 new.deferred_group = self.deferred_group 

598 new.deferred_raiseload = self.deferred_raiseload 

599 new.foreign_keys = new.column.foreign_keys 

600 new.active_history = self.active_history 

601 new._has_nullable = self._has_nullable 

602 new._attribute_options = self._attribute_options 

603 new._has_insert_default = self._has_insert_default 

604 new._has_dataclass_arguments = self._has_dataclass_arguments 

605 new._use_existing_column = self._use_existing_column 

606 new._sort_order = self._sort_order 

607 util.set_creation_order(new) 

608 return new 

609 

610 @property 

611 def name(self) -> str: 

612 return self.column.name 

613 

614 @property 

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

616 effective_deferred = self.deferred 

617 if effective_deferred is _NoArg.NO_ARG: 

618 effective_deferred = bool( 

619 self.deferred_group or self.deferred_raiseload 

620 ) 

621 

622 if effective_deferred or self.active_history: 

623 return ColumnProperty( 

624 self.column, 

625 deferred=effective_deferred, 

626 group=self.deferred_group, 

627 raiseload=self.deferred_raiseload, 

628 attribute_options=self._attribute_options, 

629 active_history=self.active_history, 

630 ) 

631 else: 

632 return None 

633 

634 @property 

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

636 return [ 

637 ( 

638 self.column, 

639 ( 

640 self._sort_order 

641 if self._sort_order is not _NoArg.NO_ARG 

642 else 0 

643 ), 

644 ) 

645 ] 

646 

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

648 return self.column 

649 

650 def operate( 

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

652 ) -> ColumnElement[Any]: 

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

654 

655 def reverse_operate( 

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

657 ) -> ColumnElement[Any]: 

658 col = self.__clause_element__() 

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

660 

661 def found_in_pep593_annotated(self) -> Any: 

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

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

664 return MappedColumn() 

665 

666 def _adjust_for_existing_column( 

667 self, 

668 decl_scan: _ClassScanMapperConfig, 

669 key: str, 

670 given_column: Column[_T], 

671 ) -> Column[_T]: 

672 if ( 

673 self._use_existing_column 

674 and decl_scan.inherits 

675 and decl_scan.single 

676 ): 

677 if decl_scan.is_deferred: 

678 raise sa_exc.ArgumentError( 

679 "Can't use use_existing_column with deferred mappers" 

680 ) 

681 supercls_mapper = class_mapper(decl_scan.inherits, False) 

682 

683 colname = ( 

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

685 ) 

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

687 colname, given_column 

688 ) 

689 return given_column 

690 

691 def declarative_scan( 

692 self, 

693 decl_scan: _ClassScanMapperConfig, 

694 registry: _RegistryType, 

695 cls: Type[Any], 

696 originating_module: Optional[str], 

697 key: str, 

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

699 annotation: Optional[_AnnotationScanType], 

700 extracted_mapped_annotation: Optional[_AnnotationScanType], 

701 is_dataclass_field: bool, 

702 ) -> None: 

703 column = self.column 

704 

705 column = self.column = self._adjust_for_existing_column( 

706 decl_scan, key, self.column 

707 ) 

708 

709 if column.key is None: 

710 column.key = key 

711 if column.name is None: 

712 column.name = key 

713 

714 sqltype = column.type 

715 

716 if extracted_mapped_annotation is None: 

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

718 self._raise_for_required(key, cls) 

719 else: 

720 return 

721 

722 self._init_column_for_annotation( 

723 cls, 

724 decl_scan, 

725 key, 

726 registry, 

727 extracted_mapped_annotation, 

728 originating_module, 

729 ) 

730 

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

732 def declarative_scan_for_composite( 

733 self, 

734 decl_scan: _ClassScanMapperConfig, 

735 registry: _RegistryType, 

736 cls: Type[Any], 

737 originating_module: Optional[str], 

738 key: str, 

739 param_name: str, 

740 param_annotation: _AnnotationScanType, 

741 ) -> None: 

742 decl_base = util.preloaded.orm_decl_base 

743 decl_base._undefer_column_name(param_name, self.column) 

744 self._init_column_for_annotation( 

745 cls, decl_scan, key, registry, param_annotation, originating_module 

746 ) 

747 

748 def _init_column_for_annotation( 

749 self, 

750 cls: Type[Any], 

751 decl_scan: _ClassScanMapperConfig, 

752 key: str, 

753 registry: _RegistryType, 

754 argument: _AnnotationScanType, 

755 originating_module: Optional[str], 

756 ) -> None: 

757 sqltype = self.column.type 

758 

759 if is_fwd_ref( 

760 argument, check_generic=True, check_for_plain_string=True 

761 ): 

762 assert originating_module is not None 

763 argument = de_stringify_annotation( 

764 cls, argument, originating_module, include_generic=True 

765 ) 

766 

767 nullable = includes_none(argument) 

768 

769 if not self._has_nullable: 

770 self.column.nullable = nullable 

771 

772 our_type = de_optionalize_union_types(argument) 

773 

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

775 our_type_is_pep593 = False 

776 raw_pep_593_type = None 

777 

778 if is_pep593(our_type): 

779 our_type_is_pep593 = True 

780 

781 pep_593_components = get_args(our_type) 

782 raw_pep_593_type = pep_593_components[0] 

783 if nullable: 

784 raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) 

785 find_mapped_in = pep_593_components[1:] 

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

787 # do not support nested annotation inside unions ets 

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

789 

790 use_args_from: Optional[MappedColumn[Any]] 

791 for elem in find_mapped_in: 

792 if isinstance(elem, MappedColumn): 

793 use_args_from = elem 

794 break 

795 else: 

796 use_args_from = None 

797 

798 if use_args_from is not None: 

799 

800 self.column = use_args_from._adjust_for_existing_column( 

801 decl_scan, key, self.column 

802 ) 

803 

804 if ( 

805 not self._has_insert_default 

806 and use_args_from.column.default is not None 

807 ): 

808 self.column.default = None 

809 

810 use_args_from.column._merge(self.column) 

811 sqltype = self.column.type 

812 

813 if ( 

814 use_args_from.deferred is not _NoArg.NO_ARG 

815 and self.deferred is _NoArg.NO_ARG 

816 ): 

817 self.deferred = use_args_from.deferred 

818 

819 if ( 

820 use_args_from.deferred_group is not None 

821 and self.deferred_group is None 

822 ): 

823 self.deferred_group = use_args_from.deferred_group 

824 

825 if ( 

826 use_args_from.deferred_raiseload is not None 

827 and self.deferred_raiseload is None 

828 ): 

829 self.deferred_raiseload = use_args_from.deferred_raiseload 

830 

831 if ( 

832 use_args_from._use_existing_column 

833 and not self._use_existing_column 

834 ): 

835 self._use_existing_column = True 

836 

837 if use_args_from.active_history: 

838 self.active_history = use_args_from.active_history 

839 

840 if ( 

841 use_args_from._sort_order is not None 

842 and self._sort_order is _NoArg.NO_ARG 

843 ): 

844 self._sort_order = use_args_from._sort_order 

845 

846 if ( 

847 use_args_from.column.key is not None 

848 or use_args_from.column.name is not None 

849 ): 

850 util.warn_deprecated( 

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

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

853 "2.0.22", 

854 ) 

855 

856 if use_args_from._has_dataclass_arguments: 

857 for idx, arg in enumerate( 

858 use_args_from._attribute_options._fields 

859 ): 

860 if ( 

861 use_args_from._attribute_options[idx] 

862 is not _NoArg.NO_ARG 

863 ): 

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

865 util.warn_deprecated( 

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

867 "cannot be specified within a mapped_column() " 

868 "bundled inside of an Annotated object", 

869 "2.0.22", 

870 ) 

871 

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