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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

357 statements  

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 # type: ignore 

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 declarative_scan( 

667 self, 

668 decl_scan: _ClassScanMapperConfig, 

669 registry: _RegistryType, 

670 cls: Type[Any], 

671 originating_module: Optional[str], 

672 key: str, 

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

674 annotation: Optional[_AnnotationScanType], 

675 extracted_mapped_annotation: Optional[_AnnotationScanType], 

676 is_dataclass_field: bool, 

677 ) -> None: 

678 column = self.column 

679 

680 if ( 

681 self._use_existing_column 

682 and decl_scan.inherits 

683 and decl_scan.single 

684 ): 

685 if decl_scan.is_deferred: 

686 raise sa_exc.ArgumentError( 

687 "Can't use use_existing_column with deferred mappers" 

688 ) 

689 supercls_mapper = class_mapper(decl_scan.inherits, False) 

690 

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

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

693 colname, column 

694 ) 

695 

696 if column.key is None: 

697 column.key = key 

698 if column.name is None: 

699 column.name = key 

700 

701 sqltype = column.type 

702 

703 if extracted_mapped_annotation is None: 

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

705 self._raise_for_required(key, cls) 

706 else: 

707 return 

708 

709 self._init_column_for_annotation( 

710 cls, 

711 registry, 

712 extracted_mapped_annotation, 

713 originating_module, 

714 ) 

715 

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

717 def declarative_scan_for_composite( 

718 self, 

719 registry: _RegistryType, 

720 cls: Type[Any], 

721 originating_module: Optional[str], 

722 key: str, 

723 param_name: str, 

724 param_annotation: _AnnotationScanType, 

725 ) -> None: 

726 decl_base = util.preloaded.orm_decl_base 

727 decl_base._undefer_column_name(param_name, self.column) 

728 self._init_column_for_annotation( 

729 cls, registry, param_annotation, originating_module 

730 ) 

731 

732 def _init_column_for_annotation( 

733 self, 

734 cls: Type[Any], 

735 registry: _RegistryType, 

736 argument: _AnnotationScanType, 

737 originating_module: Optional[str], 

738 ) -> None: 

739 sqltype = self.column.type 

740 

741 if is_fwd_ref( 

742 argument, check_generic=True, check_for_plain_string=True 

743 ): 

744 assert originating_module is not None 

745 argument = de_stringify_annotation( 

746 cls, argument, originating_module, include_generic=True 

747 ) 

748 

749 nullable = includes_none(argument) 

750 

751 if not self._has_nullable: 

752 self.column.nullable = nullable 

753 

754 our_type = de_optionalize_union_types(argument) 

755 

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

757 our_type_is_pep593 = False 

758 raw_pep_593_type = None 

759 

760 if is_pep593(our_type): 

761 our_type_is_pep593 = True 

762 

763 pep_593_components = get_args(our_type) 

764 raw_pep_593_type = pep_593_components[0] 

765 if nullable: 

766 raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) 

767 find_mapped_in = pep_593_components[1:] 

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

769 # do not support nested annotation inside unions ets 

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

771 

772 use_args_from: Optional[MappedColumn[Any]] 

773 for elem in find_mapped_in: 

774 if isinstance(elem, MappedColumn): 

775 use_args_from = elem 

776 break 

777 else: 

778 use_args_from = None 

779 

780 if use_args_from is not None: 

781 if ( 

782 not self._has_insert_default 

783 and use_args_from.column.default is not None 

784 ): 

785 self.column.default = None 

786 

787 use_args_from.column._merge(self.column) 

788 sqltype = self.column.type 

789 

790 if ( 

791 use_args_from.deferred is not _NoArg.NO_ARG 

792 and self.deferred is _NoArg.NO_ARG 

793 ): 

794 self.deferred = use_args_from.deferred 

795 

796 if ( 

797 use_args_from.deferred_group is not None 

798 and self.deferred_group is None 

799 ): 

800 self.deferred_group = use_args_from.deferred_group 

801 

802 if ( 

803 use_args_from.deferred_raiseload is not None 

804 and self.deferred_raiseload is None 

805 ): 

806 self.deferred_raiseload = use_args_from.deferred_raiseload 

807 

808 if ( 

809 use_args_from._use_existing_column 

810 and not self._use_existing_column 

811 ): 

812 self._use_existing_column = True 

813 

814 if use_args_from.active_history: 

815 self.active_history = use_args_from.active_history 

816 

817 if ( 

818 use_args_from._sort_order is not None 

819 and self._sort_order is _NoArg.NO_ARG 

820 ): 

821 self._sort_order = use_args_from._sort_order 

822 

823 if ( 

824 use_args_from.column.key is not None 

825 or use_args_from.column.name is not None 

826 ): 

827 util.warn_deprecated( 

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

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

830 "2.0.22", 

831 ) 

832 

833 if use_args_from._has_dataclass_arguments: 

834 for idx, arg in enumerate( 

835 use_args_from._attribute_options._fields 

836 ): 

837 if ( 

838 use_args_from._attribute_options[idx] 

839 is not _NoArg.NO_ARG 

840 ): 

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

842 util.warn_deprecated( 

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

844 "cannot be specified within a mapped_column() " 

845 "bundled inside of an Annotated object", 

846 "2.0.22", 

847 ) 

848 

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

850 checks: List[Any] 

851 if our_type_is_pep593: 

852 checks = [our_type, raw_pep_593_type] 

853 else: 

854 checks = [our_type] 

855 

856 for check_type in checks: 

857 new_sqltype = registry._resolve_type(check_type) 

858 if new_sqltype is not None: 

859 break 

860 else: 

861 if isinstance(our_type, TypeEngine) or ( 

862 isinstance(our_type, type) 

863 and issubclass(our_type, TypeEngine) 

864 ): 

865 raise orm_exc.MappedAnnotationError( 

866 f"The type provided inside the {self.column.key!r} " 

867 "attribute Mapped annotation is the SQLAlchemy type " 

868 f"{our_type}. Expected a Python type instead" 

869 ) 

870 elif is_a_type(our_type): 

871 raise orm_exc.MappedAnnotationError( 

872 "Could not locate SQLAlchemy Core type for Python " 

873 f"type {our_type} inside the {self.column.key!r} " 

874 "attribute Mapped annotation" 

875 ) 

876 else: 

877 raise orm_exc.MappedAnnotationError( 

878 f"The object provided inside the {self.column.key!r} " 

879 "attribute Mapped annotation is not a Python type, " 

880 f"it's the object {our_type!r}. Expected a Python " 

881 "type." 

882 ) 

883 

884 self.column._set_type(new_sqltype)