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

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 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 . import strategy_options 

34from .base import _DeclarativeMapped 

35from .base import class_mapper 

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 .relationships import RelationshipProperty 

48from .util import de_stringify_annotation 

49from .. import exc as sa_exc 

50from .. import ForeignKey 

51from .. import log 

52from .. import util 

53from ..sql import coercions 

54from ..sql import roles 

55from ..sql.base import _NoArg 

56from ..sql.schema import Column 

57from ..sql.schema import SchemaConst 

58from ..sql.type_api import TypeEngine 

59from ..util.typing import de_optionalize_union_types 

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 ForwardRef 

69 

70 from ._typing import _IdentityKeyType 

71 from ._typing import _InstanceDict 

72 from ._typing import _ORMColumnExprArgument 

73 from ._typing import _RegistryType 

74 from .base import Mapped 

75 from .decl_base import _DeclarativeMapperConfig 

76 from .mapper import Mapper 

77 from .session import Session 

78 from .state import _InstallLoaderCallableProto 

79 from .state import InstanceState 

80 from ..sql._typing import _InfoType 

81 from ..sql.elements import ColumnElement 

82 from ..sql.elements import NamedColumn 

83 from ..sql.operators import OperatorType 

84 from ..util.typing import _AnnotationScanType 

85 from ..util.typing import _MatchedOnType 

86 from ..util.typing import RODescriptorReference 

87 

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

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

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

91 

92__all__ = [ 

93 "ColumnProperty", 

94 "CompositeProperty", 

95 "ConcreteInheritedProperty", 

96 "RelationshipProperty", 

97 "SynonymProperty", 

98] 

99 

100 

101@log.class_logger 

102class ColumnProperty( 

103 _DataclassDefaultsDontSet, 

104 _MapsColumns[_T], 

105 StrategizedProperty[_T], 

106 _IntrospectsAnnotations, 

107 log.Identified, 

108): 

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

110 or other column expression. 

111 

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

113 

114 """ 

115 

116 strategy_wildcard_key = strategy_options._COLUMN_TOKEN 

117 inherit_cache = True 

118 """:meta private:""" 

119 

120 _links_to_entity = False 

121 

122 columns: List[NamedColumn[Any]] 

123 

124 _is_polymorphic_discriminator: bool 

125 

126 _mapped_by_synonym: Optional[str] 

127 

128 comparator_factory: Type[PropComparator[_T]] 

129 

130 __slots__ = ( 

131 "columns", 

132 "group", 

133 "deferred", 

134 "instrument", 

135 "comparator_factory", 

136 "active_history", 

137 "expire_on_flush", 

138 "_default_scalar_value", 

139 "_creation_order", 

140 "_is_polymorphic_discriminator", 

141 "_mapped_by_synonym", 

142 "_deferred_column_loader", 

143 "_raise_column_loader", 

144 "_renders_in_subqueries", 

145 "raiseload", 

146 ) 

147 

148 def __init__( 

149 self, 

150 column: _ORMColumnExprArgument[_T], 

151 *additional_columns: _ORMColumnExprArgument[Any], 

152 attribute_options: Optional[_AttributeOptions] = None, 

153 group: Optional[str] = None, 

154 deferred: bool = False, 

155 raiseload: bool = False, 

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

157 active_history: bool = False, 

158 default_scalar_value: Any = None, 

159 expire_on_flush: bool = True, 

160 info: Optional[_InfoType] = None, 

161 doc: Optional[str] = None, 

162 _instrument: bool = True, 

163 _assume_readonly_dc_attributes: bool = False, 

164 ): 

165 super().__init__( 

166 attribute_options=attribute_options, 

167 _assume_readonly_dc_attributes=_assume_readonly_dc_attributes, 

168 ) 

169 columns = (column,) + additional_columns 

170 self.columns = [ 

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

172 ] 

173 self.group = group 

174 self.deferred = deferred 

175 self.raiseload = raiseload 

176 self.instrument = _instrument 

177 self.comparator_factory = ( 

178 comparator_factory 

179 if comparator_factory is not None 

180 else self.__class__.Comparator 

181 ) 

182 self.active_history = active_history 

183 self._default_scalar_value = default_scalar_value 

184 self.expire_on_flush = expire_on_flush 

185 

186 if info is not None: 

187 self.info.update(info) 

188 

189 if doc is not None: 

190 self.doc = doc 

191 else: 

192 for col in reversed(self.columns): 

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

194 if doc is not None: 

195 self.doc = doc 

196 break 

197 else: 

198 self.doc = None 

199 

200 util.set_creation_order(self) 

201 

202 self.strategy_key = ( 

203 ("deferred", self.deferred), 

204 ("instrument", self.instrument), 

205 ) 

206 if self.raiseload: 

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

208 

209 def declarative_scan( 

210 self, 

211 decl_scan: _DeclarativeMapperConfig, 

212 registry: _RegistryType, 

213 cls: Type[Any], 

214 originating_module: Optional[str], 

215 key: str, 

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

217 annotation: Optional[_AnnotationScanType], 

218 extracted_mapped_annotation: Optional[_AnnotationScanType], 

219 is_dataclass_field: bool, 

220 ) -> None: 

221 column = self.columns[0] 

222 if column.key is None: 

223 column.key = key 

224 if column.name is None: 

225 column.name = key 

226 

227 @property 

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

229 return self 

230 

231 @property 

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

233 # mypy doesn't care about the isinstance here 

234 return [ 

235 (c, 0) # type: ignore 

236 for c in self.columns 

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

238 ] 

239 

240 def _memoized_attr__renders_in_subqueries(self) -> bool: 

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

242 return self.strategy._have_default_expression # type: ignore 

243 

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

245 self not in self.parent._readonly_props 

246 ) 

247 

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

249 def _memoized_attr__deferred_column_loader( 

250 self, 

251 ) -> _InstallLoaderCallableProto[Any]: 

252 state = util.preloaded.orm_state 

253 strategies = util.preloaded.orm_strategies 

254 return state.InstanceState._instance_level_callable_processor( 

255 self.parent.class_manager, 

256 strategies._LoadDeferredColumns(self.key), 

257 self.key, 

258 ) 

259 

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

261 def _memoized_attr__raise_column_loader( 

262 self, 

263 ) -> _InstallLoaderCallableProto[Any]: 

264 state = util.preloaded.orm_state 

265 strategies = util.preloaded.orm_strategies 

266 return state.InstanceState._instance_level_callable_processor( 

267 self.parent.class_manager, 

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

269 self.key, 

270 ) 

271 

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

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

274 into an instrumented attribute. 

275 """ 

276 

277 return self.expression 

278 

279 @property 

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

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

282 

283 E.g.:: 

284 

285 

286 class File(Base): 

287 # ... 

288 

289 name = Column(String(64)) 

290 extension = Column(String(8)) 

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

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

293 

294 .. seealso:: 

295 

296 :ref:`mapper_column_property_sql_expressions_composed` 

297 

298 """ 

299 return self.columns[0] 

300 

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

302 if not self.instrument: 

303 return 

304 

305 attributes._register_descriptor( 

306 mapper.class_, 

307 self.key, 

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

309 parententity=mapper, 

310 doc=self.doc, 

311 ) 

312 

313 def do_init(self) -> None: 

314 super().do_init() 

315 

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

317 self.columns 

318 ): 

319 util.warn( 

320 ( 

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

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

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

324 "mapped attribute name." 

325 ) 

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

327 ) 

328 

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

330 return ColumnProperty( 

331 *self.columns, 

332 deferred=self.deferred, 

333 group=self.group, 

334 active_history=self.active_history, 

335 default_scalar_value=self._default_scalar_value, 

336 ) 

337 

338 def merge( 

339 self, 

340 session: Session, 

341 source_state: InstanceState[Any], 

342 source_dict: _InstanceDict, 

343 dest_state: InstanceState[Any], 

344 dest_dict: _InstanceDict, 

345 load: bool, 

346 _recursive: Dict[Any, object], 

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

348 ) -> None: 

349 if not self.instrument: 

350 return 

351 elif self.key in source_dict: 

352 value = source_dict[self.key] 

353 

354 if not load: 

355 dest_dict[self.key] = value 

356 else: 

357 impl = dest_state.get_impl(self.key) 

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

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

360 dest_state._expire_attributes( 

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

362 ) 

363 

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

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

366 :class:`.ColumnProperty` attributes. 

367 

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

369 overview. 

370 

371 .. seealso:: 

372 

373 :class:`.PropComparator` 

374 

375 :class:`.ColumnOperators` 

376 

377 :ref:`types_operators` 

378 

379 :attr:`.TypeEngine.comparator_factory` 

380 

381 """ 

382 

383 if not TYPE_CHECKING: 

384 # prevent pylance from being clever about slots 

385 __slots__ = "__clause_element__", "info", "expressions" 

386 

387 prop: RODescriptorReference[ColumnProperty[_PT]] 

388 

389 expressions: Sequence[NamedColumn[Any]] 

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

391 attribute, adjusted for any aliasing in progress. 

392 

393 .. seealso:: 

394 

395 :ref:`maptojoin` - usage example 

396 """ 

397 

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

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

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

401 

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

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

404 annotations to give hints to various ORM functions as to 

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

406 to the mapper's with_polymorphic selectable if one is 

407 present. 

408 

409 """ 

410 

411 pe = self._parententity 

412 annotations: Dict[str, Any] = { 

413 "entity_namespace": pe, 

414 "parententity": pe, 

415 "parentmapper": pe, 

416 "proxy_key": self.prop.key, 

417 } 

418 

419 col = column 

420 

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

422 # the column against the polymorphic selectable. 

423 # see also orm.util._orm_downgrade_polymorphic_columns 

424 # for the reverse operation. 

425 if self._parentmapper._polymorphic_adapter: 

426 mapper_local_col = col 

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

428 

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

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

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

432 annotations["adapt_column"] = mapper_local_col 

433 

434 return col._annotate(annotations)._set_propagate_attrs( 

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

436 ) 

437 

438 if TYPE_CHECKING: 

439 

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

441 

442 def _memoized_method___clause_element__( 

443 self, 

444 ) -> NamedColumn[_PT]: 

445 if self.adapter: 

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

447 else: 

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

449 

450 def _memoized_attr_info(self) -> _InfoType: 

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

452 

453 ce = self.__clause_element__() 

454 try: 

455 return ce.info # type: ignore 

456 except AttributeError: 

457 return self.prop.info 

458 

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

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

461 attribute, adjusted for any aliasing in progress. 

462 

463 """ 

464 if self.adapter: 

465 return [ 

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

467 for col in self.prop.columns 

468 ] 

469 else: 

470 return [ 

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

472 ] 

473 

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

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

476 

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

478 """ 

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

480 

481 def operate( 

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

483 ) -> ColumnElement[Any]: 

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

485 

486 def reverse_operate( 

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

488 ) -> ColumnElement[Any]: 

489 col = self.__clause_element__() 

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

491 

492 def __str__(self) -> str: 

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

494 return object.__repr__(self) 

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

496 

497 

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

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

500 

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

502 

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

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

505 

506 .. seealso:: 

507 

508 :class:`.MappedColumn` 

509 

510 """ 

511 

512 inherit_cache = True 

513 """:meta private:""" 

514 

515 

516class MappedColumn( 

517 _DataclassDefaultsDontSet, 

518 _IntrospectsAnnotations, 

519 _MapsColumns[_T], 

520 _DeclarativeMapped[_T], 

521): 

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

523 

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

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

526 configuration. 

527 

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

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

530 

531 .. versionadded:: 2.0 

532 

533 

534 """ 

535 

536 __slots__ = ( 

537 "column", 

538 "_creation_order", 

539 "_sort_order", 

540 "foreign_keys", 

541 "_has_nullable", 

542 "_has_insert_default", 

543 "deferred", 

544 "deferred_group", 

545 "deferred_raiseload", 

546 "active_history", 

547 "_default_scalar_value", 

548 "_attribute_options", 

549 "_has_dataclass_arguments", 

550 "_use_existing_column", 

551 ) 

552 

553 deferred: Union[_NoArg, bool] 

554 deferred_raiseload: bool 

555 deferred_group: Optional[str] 

556 

557 column: Column[_T] 

558 foreign_keys: Optional[Set[ForeignKey]] 

559 _attribute_options: _AttributeOptions 

560 

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

562 self._attribute_options = attr_opts = kw.pop( 

563 "attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS 

564 ) 

565 

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

567 

568 self._has_dataclass_arguments = ( 

569 attr_opts is not None 

570 and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS 

571 and any( 

572 attr_opts[i] is not _NoArg.NO_ARG 

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

574 if attr != "dataclasses_default" 

575 ) 

576 ) 

577 

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

579 self._has_insert_default = insert_default is not _NoArg.NO_ARG 

580 self._default_scalar_value = _NoArg.NO_ARG 

581 

582 if attr_opts.dataclasses_default is not _NoArg.NO_ARG: 

583 kw["default"] = attr_opts.dataclasses_default 

584 

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

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

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

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

589 

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

591 

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

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

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

595 # dataclass role 

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

597 

598 self.foreign_keys = self.column.foreign_keys 

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

600 None, 

601 SchemaConst.NULL_UNSPECIFIED, 

602 ) 

603 util.set_creation_order(self) 

604 

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

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

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

608 new.deferred = self.deferred 

609 new.deferred_group = self.deferred_group 

610 new.deferred_raiseload = self.deferred_raiseload 

611 new.foreign_keys = new.column.foreign_keys 

612 new.active_history = self.active_history 

613 new._has_nullable = self._has_nullable 

614 new._attribute_options = self._attribute_options 

615 new._has_insert_default = self._has_insert_default 

616 new._has_dataclass_arguments = self._has_dataclass_arguments 

617 new._use_existing_column = self._use_existing_column 

618 new._sort_order = self._sort_order 

619 new._default_scalar_value = self._default_scalar_value 

620 util.set_creation_order(new) 

621 return new 

622 

623 @property 

624 def name(self) -> str: 

625 return self.column.name 

626 

627 @property 

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

629 effective_deferred = self.deferred 

630 if effective_deferred is _NoArg.NO_ARG: 

631 effective_deferred = bool( 

632 self.deferred_group or self.deferred_raiseload 

633 ) 

634 

635 if ( 

636 effective_deferred 

637 or self.active_history 

638 or self._default_scalar_value is not _NoArg.NO_ARG 

639 ): 

640 return ColumnProperty( 

641 self.column, 

642 deferred=effective_deferred, 

643 group=self.deferred_group, 

644 raiseload=self.deferred_raiseload, 

645 attribute_options=self._attribute_options, 

646 active_history=self.active_history, 

647 default_scalar_value=( 

648 self._default_scalar_value 

649 if self._default_scalar_value is not _NoArg.NO_ARG 

650 else None 

651 ), 

652 ) 

653 else: 

654 return None 

655 

656 @property 

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

658 return [ 

659 ( 

660 self.column, 

661 ( 

662 self._sort_order 

663 if self._sort_order is not _NoArg.NO_ARG 

664 else 0 

665 ), 

666 ) 

667 ] 

668 

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

670 return self.column 

671 

672 def operate( 

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

674 ) -> ColumnElement[Any]: 

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

676 

677 def reverse_operate( 

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

679 ) -> ColumnElement[Any]: 

680 col = self.__clause_element__() 

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

682 

683 def found_in_pep593_annotated(self) -> Any: 

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

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

686 return MappedColumn() 

687 

688 def _adjust_for_existing_column( 

689 self, 

690 decl_scan: _DeclarativeMapperConfig, 

691 key: str, 

692 given_column: Column[_T], 

693 ) -> Column[_T]: 

694 if ( 

695 self._use_existing_column 

696 and decl_scan.inherits 

697 and decl_scan.single 

698 ): 

699 if decl_scan.is_deferred: 

700 raise sa_exc.ArgumentError( 

701 "Can't use use_existing_column with deferred mappers" 

702 ) 

703 supercls_mapper = class_mapper(decl_scan.inherits, False) 

704 

705 colname = ( 

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

707 ) 

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

709 colname, given_column 

710 ) 

711 return given_column 

712 

713 def declarative_scan( 

714 self, 

715 decl_scan: _DeclarativeMapperConfig, 

716 registry: _RegistryType, 

717 cls: Type[Any], 

718 originating_module: Optional[str], 

719 key: str, 

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

721 annotation: Optional[_AnnotationScanType], 

722 extracted_mapped_annotation: Optional[_AnnotationScanType], 

723 is_dataclass_field: bool, 

724 ) -> None: 

725 column = self.column 

726 

727 column = self.column = self._adjust_for_existing_column( 

728 decl_scan, key, self.column 

729 ) 

730 

731 if column.key is None: 

732 column.key = key 

733 if column.name is None: 

734 column.name = key 

735 

736 sqltype = column.type 

737 

738 if extracted_mapped_annotation is None: 

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

740 self._raise_for_required(key, cls) 

741 else: 

742 return 

743 

744 self._init_column_for_annotation( 

745 cls, 

746 decl_scan, 

747 key, 

748 registry, 

749 extracted_mapped_annotation, 

750 originating_module, 

751 ) 

752 

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

754 def declarative_scan_for_composite( 

755 self, 

756 decl_scan: _DeclarativeMapperConfig, 

757 registry: _RegistryType, 

758 cls: Type[Any], 

759 originating_module: Optional[str], 

760 key: str, 

761 param_name: str, 

762 param_annotation: _AnnotationScanType, 

763 ) -> None: 

764 decl_base = util.preloaded.orm_decl_base 

765 decl_base._undefer_column_name(param_name, self.column) 

766 self._init_column_for_annotation( 

767 cls, decl_scan, key, registry, param_annotation, originating_module 

768 ) 

769 

770 def _init_column_for_annotation( 

771 self, 

772 cls: Type[Any], 

773 decl_scan: _DeclarativeMapperConfig, 

774 key: str, 

775 registry: _RegistryType, 

776 argument: _AnnotationScanType, 

777 originating_module: Optional[str], 

778 ) -> None: 

779 sqltype = self.column.type 

780 

781 de_stringified_argument: _MatchedOnType 

782 

783 if is_fwd_ref( 

784 argument, check_generic=True, check_for_plain_string=True 

785 ): 

786 assert originating_module is not None 

787 de_stringified_argument = de_stringify_annotation( 

788 cls, argument, originating_module, include_generic=True 

789 ) 

790 else: 

791 if TYPE_CHECKING: 

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

793 de_stringified_argument = argument 

794 

795 nullable = includes_none(de_stringified_argument) 

796 

797 if not self._has_nullable: 

798 self.column.nullable = nullable 

799 

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

801 raw_pep_593_type = resolved_pep_593_type = None 

802 raw_pep_695_type = resolved_pep_695_type = None 

803 

804 our_type: Any = de_optionalize_union_types(de_stringified_argument) 

805 

806 if is_pep695(our_type): 

807 raw_pep_695_type = our_type 

808 our_type = de_optionalize_union_types(raw_pep_695_type.__value__) 

809 our_args = get_args(raw_pep_695_type) 

810 if our_args: 

811 our_type = our_type[our_args] 

812 

813 resolved_pep_695_type = our_type 

814 

815 if is_pep593(our_type): 

816 pep_593_components = get_args(our_type) 

817 raw_pep_593_type = our_type 

818 resolved_pep_593_type = pep_593_components[0] 

819 if nullable: 

820 resolved_pep_593_type = de_optionalize_union_types( 

821 resolved_pep_593_type 

822 ) 

823 find_mapped_in = pep_593_components[1:] 

824 

825 use_args_from: Optional[MappedColumn[Any]] 

826 for elem in find_mapped_in: 

827 if isinstance(elem, MappedColumn): 

828 use_args_from = elem 

829 break 

830 else: 

831 use_args_from = None 

832 

833 if use_args_from is not None: 

834 

835 self.column = use_args_from._adjust_for_existing_column( 

836 decl_scan, key, self.column 

837 ) 

838 

839 if ( 

840 self._has_insert_default 

841 or self._attribute_options.dataclasses_default 

842 is not _NoArg.NO_ARG 

843 ): 

844 omit_defaults = True 

845 else: 

846 omit_defaults = False 

847 

848 use_args_from.column._merge( 

849 self.column, omit_defaults=omit_defaults 

850 ) 

851 sqltype = self.column.type 

852 

853 if ( 

854 use_args_from.deferred is not _NoArg.NO_ARG 

855 and self.deferred is _NoArg.NO_ARG 

856 ): 

857 self.deferred = use_args_from.deferred 

858 

859 if ( 

860 use_args_from.deferred_group is not None 

861 and self.deferred_group is None 

862 ): 

863 self.deferred_group = use_args_from.deferred_group 

864 

865 if ( 

866 use_args_from.deferred_raiseload is not None 

867 and self.deferred_raiseload is None 

868 ): 

869 self.deferred_raiseload = use_args_from.deferred_raiseload 

870 

871 if ( 

872 use_args_from._use_existing_column 

873 and not self._use_existing_column 

874 ): 

875 self._use_existing_column = True