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

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

358 statements  

1# orm/properties.py 

2# Copyright (C) 2005-2024 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 strategy_options 

32from .base import _DeclarativeMapped 

33from .base import class_mapper 

34from .descriptor_props import CompositeProperty 

35from .descriptor_props import ConcreteInheritedProperty 

36from .descriptor_props import SynonymProperty 

37from .interfaces import _AttributeOptions 

38from .interfaces import _DEFAULT_ATTRIBUTE_OPTIONS 

39from .interfaces import _IntrospectsAnnotations 

40from .interfaces import _MapsColumns 

41from .interfaces import MapperProperty 

42from .interfaces import PropComparator 

43from .interfaces import StrategizedProperty 

44from .relationships import RelationshipProperty 

45from .util import de_stringify_annotation 

46from .util import de_stringify_union_elements 

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 is_fwd_ref 

59from ..util.typing import is_optional_union 

60from ..util.typing import is_pep593 

61from ..util.typing import is_pep695 

62from ..util.typing import is_union 

63from ..util.typing import Self 

64from ..util.typing import typing_get_args 

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 isinstance(argument, str) or is_fwd_ref( 

742 argument, check_generic=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 if is_union(argument): 

750 assert originating_module is not None 

751 argument = de_stringify_union_elements( 

752 cls, argument, originating_module 

753 ) 

754 

755 nullable = is_optional_union(argument) 

756 

757 if not self._has_nullable: 

758 self.column.nullable = nullable 

759 

760 our_type = de_optionalize_union_types(argument) 

761 

762 use_args_from = None 

763 

764 our_original_type = our_type 

765 

766 if is_pep695(our_type): 

767 our_type = our_type.__value__ 

768 

769 if is_pep593(our_type): 

770 our_type_is_pep593 = True 

771 

772 pep_593_components = typing_get_args(our_type) 

773 raw_pep_593_type = pep_593_components[0] 

774 if is_optional_union(raw_pep_593_type): 

775 raw_pep_593_type = de_optionalize_union_types(raw_pep_593_type) 

776 

777 nullable = True 

778 if not self._has_nullable: 

779 self.column.nullable = nullable 

780 for elem in pep_593_components[1:]: 

781 if isinstance(elem, MappedColumn): 

782 use_args_from = elem 

783 break 

784 else: 

785 our_type_is_pep593 = False 

786 raw_pep_593_type = None 

787 

788 if use_args_from is not None: 

789 if ( 

790 not self._has_insert_default 

791 and use_args_from.column.default is not None 

792 ): 

793 self.column.default = None 

794 

795 use_args_from.column._merge(self.column) 

796 sqltype = self.column.type 

797 

798 if ( 

799 use_args_from.deferred is not _NoArg.NO_ARG 

800 and self.deferred is _NoArg.NO_ARG 

801 ): 

802 self.deferred = use_args_from.deferred 

803 

804 if ( 

805 use_args_from.deferred_group is not None 

806 and self.deferred_group is None 

807 ): 

808 self.deferred_group = use_args_from.deferred_group 

809 

810 if ( 

811 use_args_from.deferred_raiseload is not None 

812 and self.deferred_raiseload is None 

813 ): 

814 self.deferred_raiseload = use_args_from.deferred_raiseload 

815 

816 if ( 

817 use_args_from._use_existing_column 

818 and not self._use_existing_column 

819 ): 

820 self._use_existing_column = True 

821 

822 if use_args_from.active_history: 

823 self.active_history = use_args_from.active_history 

824 

825 if ( 

826 use_args_from._sort_order is not None 

827 and self._sort_order is _NoArg.NO_ARG 

828 ): 

829 self._sort_order = use_args_from._sort_order 

830 

831 if ( 

832 use_args_from.column.key is not None 

833 or use_args_from.column.name is not None 

834 ): 

835 util.warn_deprecated( 

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

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

838 "2.0.22", 

839 ) 

840 

841 if use_args_from._has_dataclass_arguments: 

842 for idx, arg in enumerate( 

843 use_args_from._attribute_options._fields 

844 ): 

845 if ( 

846 use_args_from._attribute_options[idx] 

847 is not _NoArg.NO_ARG 

848 ): 

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

850 util.warn_deprecated( 

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

852 "cannot be specified within a mapped_column() " 

853 "bundled inside of an Annotated object", 

854 "2.0.22", 

855 ) 

856 

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

858 new_sqltype = None 

859 

860 if our_type_is_pep593: 

861 checks = [our_original_type, raw_pep_593_type] 

862 else: 

863 checks = [our_original_type] 

864 

865 for check_type in checks: 

866 new_sqltype = registry._resolve_type(check_type) 

867 if new_sqltype is not None: 

868 break 

869 else: 

870 if isinstance(our_type, TypeEngine) or ( 

871 isinstance(our_type, type) 

872 and issubclass(our_type, TypeEngine) 

873 ): 

874 raise sa_exc.ArgumentError( 

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

876 "attribute Mapped annotation is the SQLAlchemy type " 

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

878 ) 

879 else: 

880 raise sa_exc.ArgumentError( 

881 "Could not locate SQLAlchemy Core type for Python " 

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

883 "attribute Mapped annotation" 

884 ) 

885 

886 self.column._set_type(new_sqltype)