Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/strategy

1# orm/strategy_options.py

3# <see AUTHORS file>

5# This module is part of SQLAlchemy and is released under

6# the MIT License: https://www.opensource.org/licenses/mit-license.php

7# mypy: allow-untyped-defs, allow-untyped-calls

9""" """

11from __future__ import annotations

13import typing

14from typing import Any

15from typing import Callable

16from typing import cast

17from typing import Dict

18from typing import Iterable

19from typing import Optional

20from typing import overload

21from typing import Sequence

22from typing import Tuple

23from typing import Type

24from typing import TypeVar

25from typing import Union

27from . import util as orm_util

28from ._typing import insp_is_aliased_class

29from ._typing import insp_is_attribute

30from ._typing import insp_is_mapper

31from ._typing import insp_is_mapper_property

32from .attributes import QueryableAttribute

33from .base import InspectionAttr

34from .interfaces import LoaderOption

35from .path_registry import _ACCEPTED_TOKENS

36from .path_registry import _COLUMN_TOKEN

37from .path_registry import _DEFAULT_TOKEN

38from .path_registry import _RELATIONSHIP_TOKEN

39from .path_registry import _StrPathToken

40from .path_registry import _WILDCARD_TOKEN

41from .path_registry import AbstractEntityRegistry

42from .path_registry import path_is_property

43from .path_registry import PathRegistry

44from .path_registry import TokenRegistry

45from .util import _orm_full_deannotate

46from .util import AliasedInsp

47from .. import exc as sa_exc

48from .. import inspect

49from .. import util

50from ..sql import and_

51from ..sql import cache_key

52from ..sql import coercions

53from ..sql import roles

54from ..sql import traversals

55from ..sql import visitors

56from ..sql.base import _generative

57from ..util.typing import Literal

58from ..util.typing import Self

60_FN = TypeVar("_FN", bound="Callable[..., Any]")

62if typing.TYPE_CHECKING:

63 from ._typing import _EntityType

64 from ._typing import _InternalEntityType

65 from .context import _MapperEntity

66 from .context import ORMCompileState

67 from .context import QueryContext

68 from .interfaces import _StrategyKey

69 from .interfaces import MapperProperty

70 from .interfaces import ORMOption

71 from .mapper import Mapper

72 from .path_registry import _PathRepresentation

73 from ..sql._typing import _ColumnExpressionArgument

74 from ..sql._typing import _FromClauseArgument

75 from ..sql.cache_key import _CacheKeyTraversalType

76 from ..sql.cache_key import CacheKey

79_AttrType = Union[Literal["*"], "QueryableAttribute[Any]"]

81_WildcardKeyType = Literal["relationship", "column"]

82_StrategySpec = Dict[str, Any]

83_OptsType = Dict[str, Any]

84_AttrGroupType = Tuple[_AttrType, ...]

87class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):

88 __slots__ = ("propagate_to_loaders",)

90 _is_strategy_option = True

91 propagate_to_loaders: bool

93 def contains_eager(

94 self,

95 attr: _AttrType,

96 alias: Optional[_FromClauseArgument] = None,

97 _is_chain: bool = False,

98 _propagate_to_loaders: bool = False,

99 ) -> Self:

100 r"""Indicate that the given attribute should be eagerly loaded from

101 columns stated manually in the query.

102

103 This function is part of the :class:`_orm.Load` interface and supports

104 both method-chained and standalone operation.

105

106 The option is used in conjunction with an explicit join that loads

107 the desired rows, i.e.::

108

109 sess.query(Order).join(Order.user).options(contains_eager(Order.user))

110

111 The above query would join from the ``Order`` entity to its related

112 ``User`` entity, and the returned ``Order`` objects would have the

113 ``Order.user`` attribute pre-populated.

114

115 It may also be used for customizing the entries in an eagerly loaded

116 collection; queries will normally want to use the

117 :ref:`orm_queryguide_populate_existing` execution option assuming the

118 primary collection of parent objects may already have been loaded::

119

120 sess.query(User).join(User.addresses).filter(

121 Address.email_address.like("%@aol.com")

122 ).options(contains_eager(User.addresses)).populate_existing()

123

124 See the section :ref:`contains_eager` for complete usage details.

125

126 .. seealso::

127

128 :ref:`loading_toplevel`

129

130 :ref:`contains_eager`

131

132 """

133 if alias is not None:

134 if not isinstance(alias, str):

135 coerced_alias = coercions.expect(roles.FromClauseRole, alias)

136 else:

137 util.warn_deprecated(

138 "Passing a string name for the 'alias' argument to "

139 "'contains_eager()` is deprecated, and will not work in a "

140 "future release. Please use a sqlalchemy.alias() or "

141 "sqlalchemy.orm.aliased() construct.",

142 version="1.4",

143 )

144 coerced_alias = alias

145

146 elif getattr(attr, "_of_type", None):

147 assert isinstance(attr, QueryableAttribute)

148 ot: Optional[_InternalEntityType[Any]] = inspect(attr._of_type)

149 assert ot is not None

150 coerced_alias = ot.selectable

151 else:

152 coerced_alias = None

153

154 cloned = self._set_relationship_strategy(

155 attr,

156 {"lazy": "joined"},

157 propagate_to_loaders=_propagate_to_loaders,

158 opts={"eager_from_alias": coerced_alias},

159 _reconcile_to_other=True if _is_chain else None,

160 )

161 return cloned

162

163 def load_only(self, *attrs: _AttrType, raiseload: bool = False) -> Self:

164 r"""Indicate that for a particular entity, only the given list

165 of column-based attribute names should be loaded; all others will be

166 deferred.

167

168 This function is part of the :class:`_orm.Load` interface and supports

169 both method-chained and standalone operation.

170

171 Example - given a class ``User``, load only the ``name`` and

172 ``fullname`` attributes::

173

174 session.query(User).options(load_only(User.name, User.fullname))

175

176 Example - given a relationship ``User.addresses -> Address``, specify

177 subquery loading for the ``User.addresses`` collection, but on each

178 ``Address`` object load only the ``email_address`` attribute::

179

180 session.query(User).options(

181 subqueryload(User.addresses).load_only(Address.email_address)

182 )

183

184 For a statement that has multiple entities,

185 the lead entity can be

186 specifically referred to using the :class:`_orm.Load` constructor::

187

188 stmt = (

189 select(User, Address)

190 .join(User.addresses)

191 .options(

192 Load(User).load_only(User.name, User.fullname),

193 Load(Address).load_only(Address.email_address),

194 )

195 )

196

197 When used together with the

198 :ref:`populate_existing <orm_queryguide_populate_existing>`

199 execution option only the attributes listed will be refreshed.

200

201 :param \*attrs: Attributes to be loaded, all others will be deferred.

202

203 :param raiseload: raise :class:`.InvalidRequestError` rather than

204 lazy loading a value when a deferred attribute is accessed. Used

205 to prevent unwanted SQL from being emitted.

206

207 .. versionadded:: 2.0

208

209 .. seealso::

210

211 :ref:`orm_queryguide_column_deferral` - in the

212 :ref:`queryguide_toplevel`

213

214 :param \*attrs: Attributes to be loaded, all others will be deferred.

215

216 :param raiseload: raise :class:`.InvalidRequestError` rather than

217 lazy loading a value when a deferred attribute is accessed. Used

218 to prevent unwanted SQL from being emitted.

219

220 .. versionadded:: 2.0

221

222 """

223 cloned = self._set_column_strategy(

224 _expand_column_strategy_attrs(attrs),

225 {"deferred": False, "instrument": True},

226 )

227

228 wildcard_strategy = {"deferred": True, "instrument": True}

229 if raiseload:

230 wildcard_strategy["raiseload"] = True

231

232 cloned = cloned._set_column_strategy(

233 ("*",),

234 wildcard_strategy,

235 )

236 return cloned

237

238 def joinedload(

239 self,

240 attr: _AttrType,

241 innerjoin: Optional[bool] = None,

242 ) -> Self:

243 """Indicate that the given attribute should be loaded using joined

244 eager loading.

245

246 This function is part of the :class:`_orm.Load` interface and supports

247 both method-chained and standalone operation.

248

249 examples::

250

251 # joined-load the "orders" collection on "User"

252 select(User).options(joinedload(User.orders))

253

254 # joined-load Order.items and then Item.keywords

255 select(Order).options(joinedload(Order.items).joinedload(Item.keywords))

256

257 # lazily load Order.items, but when Items are loaded,

258 # joined-load the keywords collection

259 select(Order).options(lazyload(Order.items).joinedload(Item.keywords))

260

261 :param innerjoin: if ``True``, indicates that the joined eager load

262 should use an inner join instead of the default of left outer join::

263

264 select(Order).options(joinedload(Order.user, innerjoin=True))

265

266 In order to chain multiple eager joins together where some may be

267 OUTER and others INNER, right-nested joins are used to link them::

268

269 select(A).options(

270 joinedload(A.bs, innerjoin=False).joinedload(B.cs, innerjoin=True)

271 )

272

273 The above query, linking A.bs via "outer" join and B.cs via "inner"

274 join would render the joins as "a LEFT OUTER JOIN (b JOIN c)". When

275 using older versions of SQLite (< 3.7.16), this form of JOIN is

276 translated to use full subqueries as this syntax is otherwise not

277 directly supported.

278

279 The ``innerjoin`` flag can also be stated with the term ``"unnested"``.

280 This indicates that an INNER JOIN should be used, *unless* the join

281 is linked to a LEFT OUTER JOIN to the left, in which case it

282 will render as LEFT OUTER JOIN. For example, supposing ``A.bs``

283 is an outerjoin::

284

285 select(A).options(joinedload(A.bs).joinedload(B.cs, innerjoin="unnested"))

286

287 The above join will render as "a LEFT OUTER JOIN b LEFT OUTER JOIN c",

288 rather than as "a LEFT OUTER JOIN (b JOIN c)".

289

290 .. note:: The "unnested" flag does **not** affect the JOIN rendered

291 from a many-to-many association table, e.g. a table configured as

292 :paramref:`_orm.relationship.secondary`, to the target table; for

293 correctness of results, these joins are always INNER and are

294 therefore right-nested if linked to an OUTER join.

295

296 .. note::

297

298 The joins produced by :func:`_orm.joinedload` are **anonymously

299 aliased**. The criteria by which the join proceeds cannot be

300 modified, nor can the ORM-enabled :class:`_sql.Select` or legacy

301 :class:`_query.Query` refer to these joins in any way, including

302 ordering. See :ref:`zen_of_eager_loading` for further detail.

303

304 To produce a specific SQL JOIN which is explicitly available, use

305 :meth:`_sql.Select.join` and :meth:`_query.Query.join`. To combine

306 explicit JOINs with eager loading of collections, use

307 :func:`_orm.contains_eager`; see :ref:`contains_eager`.

308

309 .. seealso::

310

311 :ref:`loading_toplevel`

312

313 :ref:`joined_eager_loading`

314

315 """ # noqa: E501

316 loader = self._set_relationship_strategy(

317 attr,

318 {"lazy": "joined"},

319 opts=(

320 {"innerjoin": innerjoin}

321 if innerjoin is not None

322 else util.EMPTY_DICT

323 ),

324 )

325 return loader

326

327 def subqueryload(self, attr: _AttrType) -> Self:

328 """Indicate that the given attribute should be loaded using

329 subquery eager loading.

330

331 This function is part of the :class:`_orm.Load` interface and supports

332 both method-chained and standalone operation.

333

334 examples::

335

336 # subquery-load the "orders" collection on "User"

337 select(User).options(subqueryload(User.orders))

338

339 # subquery-load Order.items and then Item.keywords

340 select(Order).options(

341 subqueryload(Order.items).subqueryload(Item.keywords)

342 )

343

344 # lazily load Order.items, but when Items are loaded,

345 # subquery-load the keywords collection

346 select(Order).options(lazyload(Order.items).subqueryload(Item.keywords))

347

348 .. seealso::

349

350 :ref:`loading_toplevel`

351

352 :ref:`subquery_eager_loading`

353

354 """

355 return self._set_relationship_strategy(attr, {"lazy": "subquery"})

356

357 def selectinload(

358 self,

359 attr: _AttrType,

360 recursion_depth: Optional[int] = None,

361 ) -> Self:

362 """Indicate that the given attribute should be loaded using

363 SELECT IN eager loading.

364

365 This function is part of the :class:`_orm.Load` interface and supports

366 both method-chained and standalone operation.

367

368 examples::

369

370 # selectin-load the "orders" collection on "User"

371 select(User).options(selectinload(User.orders))

372

373 # selectin-load Order.items and then Item.keywords

374 select(Order).options(

375 selectinload(Order.items).selectinload(Item.keywords)

376 )

377

378 # lazily load Order.items, but when Items are loaded,

379 # selectin-load the keywords collection

380 select(Order).options(lazyload(Order.items).selectinload(Item.keywords))

381

382 :param recursion_depth: optional int; when set to a positive integer

383 in conjunction with a self-referential relationship,

384 indicates "selectin" loading will continue that many levels deep

385 automatically until no items are found.

386

387 .. note:: The :paramref:`_orm.selectinload.recursion_depth` option

388 currently supports only self-referential relationships. There

389 is not yet an option to automatically traverse recursive structures

390 with more than one relationship involved.

391

392 Additionally, the :paramref:`_orm.selectinload.recursion_depth`

393 parameter is new and experimental and should be treated as "alpha"

394 status for the 2.0 series.

395

396 .. versionadded:: 2.0 added

397 :paramref:`_orm.selectinload.recursion_depth`

398

399

400 .. seealso::

401

402 :ref:`loading_toplevel`

403

404 :ref:`selectin_eager_loading`

405

406 """

407 return self._set_relationship_strategy(

408 attr,

409 {"lazy": "selectin"},

410 opts={"recursion_depth": recursion_depth},

411 )

412

413 def lazyload(self, attr: _AttrType) -> Self:

414 """Indicate that the given attribute should be loaded using "lazy"

415 loading.

416

417 This function is part of the :class:`_orm.Load` interface and supports

418 both method-chained and standalone operation.

419

420 .. seealso::

421

422 :ref:`loading_toplevel`

423

424 :ref:`lazy_loading`

425

426 """

427 return self._set_relationship_strategy(attr, {"lazy": "select"})

428

429 def immediateload(

430 self,

431 attr: _AttrType,

432 recursion_depth: Optional[int] = None,

433 ) -> Self:

434 """Indicate that the given attribute should be loaded using

435 an immediate load with a per-attribute SELECT statement.

436

437 The load is achieved using the "lazyloader" strategy and does not

438 fire off any additional eager loaders.

439

440 The :func:`.immediateload` option is superseded in general

441 by the :func:`.selectinload` option, which performs the same task

442 more efficiently by emitting a SELECT for all loaded objects.

443

444 This function is part of the :class:`_orm.Load` interface and supports

445 both method-chained and standalone operation.

446

447 :param recursion_depth: optional int; when set to a positive integer

448 in conjunction with a self-referential relationship,

449 indicates "selectin" loading will continue that many levels deep

450 automatically until no items are found.

451

452 .. note:: The :paramref:`_orm.immediateload.recursion_depth` option

453 currently supports only self-referential relationships. There

454 is not yet an option to automatically traverse recursive structures

455 with more than one relationship involved.

456

457 .. warning:: This parameter is new and experimental and should be

458 treated as "alpha" status

459

460 .. versionadded:: 2.0 added

461 :paramref:`_orm.immediateload.recursion_depth`

462

463

464 .. seealso::

465

466 :ref:`loading_toplevel`

467

468 :ref:`selectin_eager_loading`

469

470 """

471 loader = self._set_relationship_strategy(

472 attr,

473 {"lazy": "immediate"},

474 opts={"recursion_depth": recursion_depth},

475 )

476 return loader

477

478 def noload(self, attr: _AttrType) -> Self:

479 """Indicate that the given relationship attribute should remain

480 unloaded.

481

482 The relationship attribute will return ``None`` when accessed without

483 producing any loading effect.

484

485 This function is part of the :class:`_orm.Load` interface and supports

486 both method-chained and standalone operation.

487

488 :func:`_orm.noload` applies to :func:`_orm.relationship` attributes

489 only.

490

491 .. legacy:: The :func:`_orm.noload` option is **legacy**. As it

492 forces collections to be empty, which invariably leads to

493 non-intuitive and difficult to predict results. There are no

494 legitimate uses for this option in modern SQLAlchemy.

495

496 .. seealso::

497

498 :ref:`loading_toplevel`

499

500 """

501

502 return self._set_relationship_strategy(attr, {"lazy": "noload"})

503

504 def raiseload(self, attr: _AttrType, sql_only: bool = False) -> Self:

505 """Indicate that the given attribute should raise an error if accessed.

506

507 A relationship attribute configured with :func:`_orm.raiseload` will

508 raise an :exc:`~sqlalchemy.exc.InvalidRequestError` upon access. The

509 typical way this is useful is when an application is attempting to

510 ensure that all relationship attributes that are accessed in a

511 particular context would have been already loaded via eager loading.

512 Instead of having to read through SQL logs to ensure lazy loads aren't

513 occurring, this strategy will cause them to raise immediately.

514

515 :func:`_orm.raiseload` applies to :func:`_orm.relationship` attributes

516 only. In order to apply raise-on-SQL behavior to a column-based

517 attribute, use the :paramref:`.orm.defer.raiseload` parameter on the

518 :func:`.defer` loader option.

519

520 :param sql_only: if True, raise only if the lazy load would emit SQL,

521 but not if it is only checking the identity map, or determining that

522 the related value should just be None due to missing keys. When False,

523 the strategy will raise for all varieties of relationship loading.

524

525 This function is part of the :class:`_orm.Load` interface and supports

526 both method-chained and standalone operation.

527

528 .. seealso::

529

530 :ref:`loading_toplevel`

531

532 :ref:`prevent_lazy_with_raiseload`

533

534 :ref:`orm_queryguide_deferred_raiseload`

535

536 """

537

538 return self._set_relationship_strategy(

539 attr, {"lazy": "raise_on_sql" if sql_only else "raise"}

540 )

541

542 def defaultload(self, attr: _AttrType) -> Self:

543 """Indicate an attribute should load using its predefined loader style.

544

545 The behavior of this loading option is to not change the current

546 loading style of the attribute, meaning that the previously configured

547 one is used or, if no previous style was selected, the default

548 loading will be used.

549

550 This method is used to link to other loader options further into

551 a chain of attributes without altering the loader style of the links

552 along the chain. For example, to set joined eager loading for an

553 element of an element::

554

555 session.query(MyClass).options(

556 defaultload(MyClass.someattribute).joinedload(

557 MyOtherClass.someotherattribute

558 )

559 )

560

561 :func:`.defaultload` is also useful for setting column-level options on

562 a related class, namely that of :func:`.defer` and :func:`.undefer`::

563

564 session.scalars(

565 select(MyClass).options(

566 defaultload(MyClass.someattribute)

567 .defer("some_column")

568 .undefer("some_other_column")

569 )

570 )

571

572 .. seealso::

573

574 :ref:`orm_queryguide_relationship_sub_options`

575

576 :meth:`_orm.Load.options`

577

578 """

579 return self._set_relationship_strategy(attr, None)

580

581 def defer(self, key: _AttrType, raiseload: bool = False) -> Self:

582 r"""Indicate that the given column-oriented attribute should be

583 deferred, e.g. not loaded until accessed.

584

585 This function is part of the :class:`_orm.Load` interface and supports

586 both method-chained and standalone operation.

587

588 e.g.::

589

590 from sqlalchemy.orm import defer

591

592 session.query(MyClass).options(

593 defer(MyClass.attribute_one), defer(MyClass.attribute_two)

594 )

595

596 To specify a deferred load of an attribute on a related class,

597 the path can be specified one token at a time, specifying the loading

598 style for each link along the chain. To leave the loading style

599 for a link unchanged, use :func:`_orm.defaultload`::

600

601 session.query(MyClass).options(

602 defaultload(MyClass.someattr).defer(RelatedClass.some_column)

603 )

604

605 Multiple deferral options related to a relationship can be bundled

606 at once using :meth:`_orm.Load.options`::

607

608

609 select(MyClass).options(

610 defaultload(MyClass.someattr).options(

611 defer(RelatedClass.some_column),

612 defer(RelatedClass.some_other_column),

613 defer(RelatedClass.another_column),

614 )

615 )

616

617 :param key: Attribute to be deferred.

618

619 :param raiseload: raise :class:`.InvalidRequestError` rather than

620 lazy loading a value when the deferred attribute is accessed. Used

621 to prevent unwanted SQL from being emitted.

622

623 .. versionadded:: 1.4

624

625 .. seealso::

626

627 :ref:`orm_queryguide_column_deferral` - in the

628 :ref:`queryguide_toplevel`

629

630 :func:`_orm.load_only`

631

632 :func:`_orm.undefer`

633

634 """

635 strategy = {"deferred": True, "instrument": True}

636 if raiseload:

637 strategy["raiseload"] = True

638 return self._set_column_strategy(

639 _expand_column_strategy_attrs((key,)), strategy

640 )

641

642 def undefer(self, key: _AttrType) -> Self:

643 r"""Indicate that the given column-oriented attribute should be

644 undeferred, e.g. specified within the SELECT statement of the entity

645 as a whole.

646

647 The column being undeferred is typically set up on the mapping as a

648 :func:`.deferred` attribute.

649

650 This function is part of the :class:`_orm.Load` interface and supports

651 both method-chained and standalone operation.

652

653 Examples::

654

655 # undefer two columns

656 session.query(MyClass).options(

657 undefer(MyClass.col1), undefer(MyClass.col2)

658 )

659

660 # undefer all columns specific to a single class using Load + *

661 session.query(MyClass, MyOtherClass).options(Load(MyClass).undefer("*"))

662

663 # undefer a column on a related object

664 select(MyClass).options(defaultload(MyClass.items).undefer(MyClass.text))

665

666 :param key: Attribute to be undeferred.

667

668 .. seealso::

669

670 :ref:`orm_queryguide_column_deferral` - in the

671 :ref:`queryguide_toplevel`

672

673 :func:`_orm.defer`

674

675 :func:`_orm.undefer_group`

676

677 """ # noqa: E501

678 return self._set_column_strategy(

679 _expand_column_strategy_attrs((key,)),

680 {"deferred": False, "instrument": True},

681 )

682

683 def undefer_group(self, name: str) -> Self:

684 """Indicate that columns within the given deferred group name should be

685 undeferred.

686

687 The columns being undeferred are set up on the mapping as

688 :func:`.deferred` attributes and include a "group" name.

689

690 E.g::

691

692 session.query(MyClass).options(undefer_group("large_attrs"))

693

694 To undefer a group of attributes on a related entity, the path can be

695 spelled out using relationship loader options, such as

696 :func:`_orm.defaultload`::

697

698 select(MyClass).options(

699 defaultload("someattr").undefer_group("large_attrs")

700 )

701

702 .. seealso::

703

704 :ref:`orm_queryguide_column_deferral` - in the

705 :ref:`queryguide_toplevel`

706

707 :func:`_orm.defer`

708

709 :func:`_orm.undefer`

710

711 """

712 return self._set_column_strategy(

713 (_WILDCARD_TOKEN,), None, {f"undefer_group_{name}": True}

714 )

715

716 def with_expression(

717 self,

718 key: _AttrType,

719 expression: _ColumnExpressionArgument[Any],

720 ) -> Self:

721 r"""Apply an ad-hoc SQL expression to a "deferred expression"

722 attribute.

723

724 This option is used in conjunction with the

725 :func:`_orm.query_expression` mapper-level construct that indicates an

726 attribute which should be the target of an ad-hoc SQL expression.

727

728 E.g.::

729

730 stmt = select(SomeClass).options(

731 with_expression(SomeClass.x_y_expr, SomeClass.x + SomeClass.y)

732 )

733

734 .. versionadded:: 1.2

735

736 :param key: Attribute to be populated

737

738 :param expr: SQL expression to be applied to the attribute.

739

740 .. seealso::

741

742 :ref:`orm_queryguide_with_expression` - background and usage

743 examples

744

745 """

746

747 expression = _orm_full_deannotate(

748 coercions.expect(roles.LabeledColumnExprRole, expression)

749 )

750

751 return self._set_column_strategy(

752 (key,), {"query_expression": True}, extra_criteria=(expression,)

753 )

754

755 def selectin_polymorphic(self, classes: Iterable[Type[Any]]) -> Self:

756 """Indicate an eager load should take place for all attributes

757 specific to a subclass.

758

759 This uses an additional SELECT with IN against all matched primary

760 key values, and is the per-query analogue to the ``"selectin"``

761 setting on the :paramref:`.mapper.polymorphic_load` parameter.

762

763 .. versionadded:: 1.2

764

765 .. seealso::

766

767 :ref:`polymorphic_selectin`

768

769 """

770 self = self._set_class_strategy(

771 {"selectinload_polymorphic": True},

772 opts={

773 "entities": tuple(

774 sorted((inspect(cls) for cls in classes), key=id)

775 )

776 },

777 )

778 return self

779

780 @overload

781 def _coerce_strat(self, strategy: _StrategySpec) -> _StrategyKey: ...

782

783 @overload

784 def _coerce_strat(self, strategy: Literal[None]) -> None: ...

785

786 def _coerce_strat(

787 self, strategy: Optional[_StrategySpec]

788 ) -> Optional[_StrategyKey]:

789 if strategy is not None:

790 strategy_key = tuple(sorted(strategy.items()))

791 else:

792 strategy_key = None

793 return strategy_key

794

795 @_generative

796 def _set_relationship_strategy(

797 self,

798 attr: _AttrType,

799 strategy: Optional[_StrategySpec],

800 propagate_to_loaders: bool = True,

801 opts: Optional[_OptsType] = None,

802 _reconcile_to_other: Optional[bool] = None,

803 ) -> Self:

804 strategy_key = self._coerce_strat(strategy)

805

806 self._clone_for_bind_strategy(

807 (attr,),

808 strategy_key,

809 _RELATIONSHIP_TOKEN,

810 opts=opts,

811 propagate_to_loaders=propagate_to_loaders,

812 reconcile_to_other=_reconcile_to_other,

813 )

814 return self

815

816 @_generative

817 def _set_column_strategy(

818 self,

819 attrs: Tuple[_AttrType, ...],

820 strategy: Optional[_StrategySpec],

821 opts: Optional[_OptsType] = None,

822 extra_criteria: Optional[Tuple[Any, ...]] = None,

823 ) -> Self:

824 strategy_key = self._coerce_strat(strategy)

825

826 self._clone_for_bind_strategy(

827 attrs,

828 strategy_key,

829 _COLUMN_TOKEN,

830 opts=opts,

831 attr_group=attrs,

832 extra_criteria=extra_criteria,

833 )

834 return self

835

836 @_generative

837 def _set_generic_strategy(

838 self,

839 attrs: Tuple[_AttrType, ...],

840 strategy: _StrategySpec,

841 _reconcile_to_other: Optional[bool] = None,

842 ) -> Self:

843 strategy_key = self._coerce_strat(strategy)

844 self._clone_for_bind_strategy(

845 attrs,

846 strategy_key,

847 None,

848 propagate_to_loaders=True,

849 reconcile_to_other=_reconcile_to_other,

850 )

851 return self

852

853 @_generative

854 def _set_class_strategy(

855 self, strategy: _StrategySpec, opts: _OptsType

856 ) -> Self:

857 strategy_key = self._coerce_strat(strategy)

858

859 self._clone_for_bind_strategy(None, strategy_key, None, opts=opts)

860 return self

861

862 def _apply_to_parent(self, parent: Load) -> None:

863 """apply this :class:`_orm._AbstractLoad` object as a sub-option o

864 a :class:`_orm.Load` object.

865

866 Implementation is provided by subclasses.

867

868 """

869 raise NotImplementedError()

870

871 def options(self, *opts: _AbstractLoad) -> Self:

872 r"""Apply a series of options as sub-options to this

873 :class:`_orm._AbstractLoad` object.

874

875 Implementation is provided by subclasses.

876

877 """

878 raise NotImplementedError()

879

880 def _clone_for_bind_strategy(

881 self,

882 attrs: Optional[Tuple[_AttrType, ...]],

883 strategy: Optional[_StrategyKey],

884 wildcard_key: Optional[_WildcardKeyType],

885 opts: Optional[_OptsType] = None,

886 attr_group: Optional[_AttrGroupType] = None,

887 propagate_to_loaders: bool = True,

888 reconcile_to_other: Optional[bool] = None,

889 extra_criteria: Optional[Tuple[Any, ...]] = None,

890 ) -> Self:

891 raise NotImplementedError()

892

893 def process_compile_state_replaced_entities(

894 self,

895 compile_state: ORMCompileState,

896 mapper_entities: Sequence[_MapperEntity],

897 ) -> None:

898 if not compile_state.compile_options._enable_eagerloads:

899 return

900

901 # process is being run here so that the options given are validated

902 # against what the lead entities were, as well as to accommodate

903 # for the entities having been replaced with equivalents

904 self._process(

905 compile_state,

906 mapper_entities,

907 not bool(compile_state.current_path),

908 )

909

910 def process_compile_state(self, compile_state: ORMCompileState) -> None:

911 if not compile_state.compile_options._enable_eagerloads:

912 return

913

914 self._process(

915 compile_state,

916 compile_state._lead_mapper_entities,

917 not bool(compile_state.current_path)

918 and not compile_state.compile_options._for_refresh_state,

919 )

920

921 def _process(

922 self,

923 compile_state: ORMCompileState,

924 mapper_entities: Sequence[_MapperEntity],

925 raiseerr: bool,

926 ) -> None:

927 """implemented by subclasses"""

928 raise NotImplementedError()

929

930 @classmethod

931 def _chop_path(

932 cls,

933 to_chop: _PathRepresentation,

934 path: PathRegistry,

935 debug: bool = False,

936 ) -> Optional[_PathRepresentation]:

937 i = -1

938

939 for i, (c_token, p_token) in enumerate(

940 zip(to_chop, path.natural_path)

941 ):

942 if isinstance(c_token, str):

943 if i == 0 and (

944 c_token.endswith(f":{_DEFAULT_TOKEN}")

945 or c_token.endswith(f":{_WILDCARD_TOKEN}")

946 ):

947 return to_chop

948 elif (

949 c_token != f"{_RELATIONSHIP_TOKEN}:{_WILDCARD_TOKEN}"

950 and c_token != p_token.key # type: ignore

951 ):

952 return None

953

954 if c_token is p_token:

955 continue

956 elif (

957 isinstance(c_token, InspectionAttr)

958 and insp_is_mapper(c_token)

959 and insp_is_mapper(p_token)

960 and c_token.isa(p_token)

961 ):

962 continue

963

964 else:

965 return None

966 return to_chop[i + 1 :]

967

968

969class Load(_AbstractLoad):

970 """Represents loader options which modify the state of a

971 ORM-enabled :class:`_sql.Select` or a legacy :class:`_query.Query` in

972 order to affect how various mapped attributes are loaded.

973

974 The :class:`_orm.Load` object is in most cases used implicitly behind the

975 scenes when one makes use of a query option like :func:`_orm.joinedload`,

976 :func:`_orm.defer`, or similar. It typically is not instantiated directly

977 except for in some very specific cases.

978

979 .. seealso::

980

981 :ref:`orm_queryguide_relationship_per_entity_wildcard` - illustrates an

982 example where direct use of :class:`_orm.Load` may be useful

983

984 """

985

986 __slots__ = (

987 "path",

988 "context",

989 "additional_source_entities",

990 )

991

992 _traverse_internals = [

993 ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key),

994 (

995 "context",

996 visitors.InternalTraversal.dp_has_cache_key_list,

997 ),

998 ("propagate_to_loaders", visitors.InternalTraversal.dp_boolean),

999 (

1000 "additional_source_entities",

1001 visitors.InternalTraversal.dp_has_cache_key_list,

1002 ),

1003 ]

1004 _cache_key_traversal = None

1005

1006 path: PathRegistry

1007 context: Tuple[_LoadElement, ...]

1008 additional_source_entities: Tuple[_InternalEntityType[Any], ...]

1009

1010 def __init__(self, entity: _EntityType[Any]):

1011 insp = cast("Union[Mapper[Any], AliasedInsp[Any]]", inspect(entity))

1012 insp._post_inspect

1013

1014 self.path = insp._path_registry

1015 self.context = ()

1016 self.propagate_to_loaders = False

1017 self.additional_source_entities = ()

1018

1019 def __str__(self) -> str:

1020 return f"Load({self.path[0]})"

1021

1022 @classmethod

1023 def _construct_for_existing_path(

1024 cls, path: AbstractEntityRegistry

1025 ) -> Load:

1026 load = cls.__new__(cls)

1027 load.path = path

1028 load.context = ()

1029 load.propagate_to_loaders = False

1030 load.additional_source_entities = ()

1031 return load

1032

1033 def _adapt_cached_option_to_uncached_option(

1034 self, context: QueryContext, uncached_opt: ORMOption

1035 ) -> ORMOption:

1036 if uncached_opt is self:

1037 return self

1038 return self._adjust_for_extra_criteria(context)

1039

1040 def _prepend_path(self, path: PathRegistry) -> Load:

1041 cloned = self._clone()

1042 cloned.context = tuple(

1043 element._prepend_path(path) for element in self.context

1044 )

1045 return cloned

1046

1047 def _adjust_for_extra_criteria(self, context: QueryContext) -> Load:

1048 """Apply the current bound parameters in a QueryContext to all

1049 occurrences "extra_criteria" stored within this ``Load`` object,

1050 returning a new instance of this ``Load`` object.

1051

1052 """

1053

1054 # avoid generating cache keys for the queries if we don't

1055 # actually have any extra_criteria options, which is the

1056 # common case

1057 for value in self.context:

1058 if value._extra_criteria:

1059 break

1060 else:

1061 return self

1062

1063 replacement_cache_key = context.user_passed_query._generate_cache_key()

1064

1065 if replacement_cache_key is None:

1066 return self

1067

1068 orig_query = context.compile_state.select_statement

1069 orig_cache_key = orig_query._generate_cache_key()

1070 assert orig_cache_key is not None

1071

1072 def process(

1073 opt: _LoadElement,

1074 replacement_cache_key: CacheKey,

1075 orig_cache_key: CacheKey,

1076 ) -> _LoadElement:

1077 cloned_opt = opt._clone()

1078

1079 cloned_opt._extra_criteria = tuple(

1080 replacement_cache_key._apply_params_to_element(

1081 orig_cache_key, crit

1082 )

1083 for crit in cloned_opt._extra_criteria

1084 )

1085

1086 return cloned_opt

1087

1088 cloned = self._clone()

1089 cloned.context = tuple(

1090 (

1091 process(value, replacement_cache_key, orig_cache_key)

1092 if value._extra_criteria

1093 else value

1094 )

1095 for value in self.context

1096 )

1097 return cloned

1098

1099 def _reconcile_query_entities_with_us(self, mapper_entities, raiseerr):

1100 """called at process time to allow adjustment of the root

1101 entity inside of _LoadElement objects.

1102

1103 """

1104 path = self.path

1105

1106 for ent in mapper_entities:

1107 ezero = ent.entity_zero

1108 if ezero and orm_util._entity_corresponds_to(

1109 # technically this can be a token also, but this is

1110 # safe to pass to _entity_corresponds_to()

1111 ezero,

1112 cast("_InternalEntityType[Any]", path[0]),

1113 ):

1114 return ezero

1115

1116 return None

1117

1118 def _process(

1119 self,

1120 compile_state: ORMCompileState,

1121 mapper_entities: Sequence[_MapperEntity],

1122 raiseerr: bool,

1123 ) -> None:

1124 reconciled_lead_entity = self._reconcile_query_entities_with_us(

1125 mapper_entities, raiseerr

1126 )

1127

1128 # if the context has a current path, this is a lazy load

1129 has_current_path = bool(compile_state.compile_options._current_path)

1130

1131 for loader in self.context:

1132 # issue #11292

1133 # historically, propagate_to_loaders was only considered at

1134 # object loading time, whether or not to carry along options

1135 # onto an object's loaded state where it would be used by lazyload.

1136 # however, the defaultload() option needs to propagate in case

1137 # its sub-options propagate_to_loaders, but its sub-options

1138 # that dont propagate should not be applied for lazy loaders.

1139 # so we check again

1140 if has_current_path and not loader.propagate_to_loaders:

1141 continue

1142 loader.process_compile_state(

1143 self,

1144 compile_state,

1145 mapper_entities,

1146 reconciled_lead_entity,

1147 raiseerr,

1148 )

1149

1150 def _apply_to_parent(self, parent: Load) -> None:

1151 """apply this :class:`_orm.Load` object as a sub-option of another

1152 :class:`_orm.Load` object.

1153

1154 This method is used by the :meth:`_orm.Load.options` method.

1155

1156 """

1157 cloned = self._generate()

1158

1159 assert cloned.propagate_to_loaders == self.propagate_to_loaders

1160

1161 if not any(

1162 orm_util._entity_corresponds_to_use_path_impl(

1163 elem, cloned.path.odd_element(0)

1164 )

1165 for elem in (parent.path.odd_element(-1),)

1166 + parent.additional_source_entities

1167 ):

1168 if len(cloned.path) > 1:

1169 attrname = cloned.path[1]

1170 parent_entity = cloned.path[0]

1171 else:

1172 attrname = cloned.path[0]

1173 parent_entity = cloned.path[0]

1174 _raise_for_does_not_link(parent.path, attrname, parent_entity)

1175

1176 cloned.path = PathRegistry.coerce(parent.path[0:-1] + cloned.path[:])

1177

1178 if self.context:

1179 cloned.context = tuple(

1180 value._prepend_path_from(parent) for value in self.context

1181 )

1182

1183 if cloned.context:

1184 parent.context += cloned.context

1185 parent.additional_source_entities += (

1186 cloned.additional_source_entities

1187 )

1188

1189 @_generative

1190 def options(self, *opts: _AbstractLoad) -> Self:

1191 r"""Apply a series of options as sub-options to this

1192 :class:`_orm.Load`

1193 object.

1194

1195 E.g.::

1196

1197 query = session.query(Author)

1198 query = query.options(

1199 joinedload(Author.book).options(

1200 load_only(Book.summary, Book.excerpt),

1201 joinedload(Book.citations).options(joinedload(Citation.author)),

1202 )

1203 )

1204

1205 :param \*opts: A series of loader option objects (ultimately

1206 :class:`_orm.Load` objects) which should be applied to the path

1207 specified by this :class:`_orm.Load` object.

1208

1209 .. versionadded:: 1.3.6

1210

1211 .. seealso::

1212

1213 :func:`.defaultload`

1214

1215 :ref:`orm_queryguide_relationship_sub_options`

1216

1217 """

1218 for opt in opts:

1219 try:

1220 opt._apply_to_parent(self)

1221 except AttributeError as ae:

1222 if not isinstance(opt, _AbstractLoad):

1223 raise sa_exc.ArgumentError(

1224 f"Loader option {opt} is not compatible with the "

1225 "Load.options() method."

1226 ) from ae

1227 else:

1228 raise

1229 return self

1230

1231 def _clone_for_bind_strategy(

1232 self,

1233 attrs: Optional[Tuple[_AttrType, ...]],

1234 strategy: Optional[_StrategyKey],

1235 wildcard_key: Optional[_WildcardKeyType],

1236 opts: Optional[_OptsType] = None,

1237 attr_group: Optional[_AttrGroupType] = None,

1238 propagate_to_loaders: bool = True,

1239 reconcile_to_other: Optional[bool] = None,

1240 extra_criteria: Optional[Tuple[Any, ...]] = None,

1241 ) -> Self:

1242 # for individual strategy that needs to propagate, set the whole

1243 # Load container to also propagate, so that it shows up in

1244 # InstanceState.load_options

1245 if propagate_to_loaders:

1246 self.propagate_to_loaders = True

1247

1248 if self.path.is_token:

1249 raise sa_exc.ArgumentError(

1250 "Wildcard token cannot be followed by another entity"

1251 )

1252

1253 elif path_is_property(self.path):

1254 # reuse the lookup which will raise a nicely formatted

1255 # LoaderStrategyException

1256 if strategy:

1257 self.path.prop._strategy_lookup(self.path.prop, strategy[0])

1258 else:

1259 raise sa_exc.ArgumentError(

1260 f"Mapped attribute '{self.path.prop}' does not "

1261 "refer to a mapped entity"

1262 )

1263

1264 if attrs is None:

1265 load_element = _ClassStrategyLoad.create(

1266 self.path,

1267 None,

1268 strategy,

1269 wildcard_key,

1270 opts,

1271 propagate_to_loaders,

1272 attr_group=attr_group,

1273 reconcile_to_other=reconcile_to_other,

1274 extra_criteria=extra_criteria,

1275 )

1276 if load_element:

1277 self.context += (load_element,)

1278 assert opts is not None

1279 self.additional_source_entities += cast(

1280 "Tuple[_InternalEntityType[Any]]", opts["entities"]

1281 )

1282

1283 else:

1284 for attr in attrs:

1285 if isinstance(attr, str):

1286 load_element = _TokenStrategyLoad.create(

1287 self.path,

1288 attr,

1289 strategy,

1290 wildcard_key,

1291 opts,

1292 propagate_to_loaders,

1293 attr_group=attr_group,

1294 reconcile_to_other=reconcile_to_other,

1295 extra_criteria=extra_criteria,

1296 )

1297 else:

1298 load_element = _AttributeStrategyLoad.create(

1299 self.path,

1300 attr,

1301 strategy,

1302 wildcard_key,

1303 opts,

1304 propagate_to_loaders,

1305 attr_group=attr_group,

1306 reconcile_to_other=reconcile_to_other,

1307 extra_criteria=extra_criteria,

1308 )

1309

1310 if load_element:

1311 # for relationship options, update self.path on this Load

1312 # object with the latest path.

1313 if wildcard_key is _RELATIONSHIP_TOKEN:

1314 self.path = load_element.path

1315 self.context += (load_element,)

1316

1317 # this seems to be effective for selectinloader,

1318 # giving the extra match to one more level deep.

1319 # but does not work for immediateloader, which still

1320 # must add additional options at load time

1321 if load_element.local_opts.get("recursion_depth", False):

1322 r1 = load_element._recurse()

1323 self.context += (r1,)

1324

1325 return self

1326

1327 def __getstate__(self):

1328 d = self._shallow_to_dict()

1329 d["path"] = self.path.serialize()

1330 return d

1331

1332 def __setstate__(self, state):

1333 state["path"] = PathRegistry.deserialize(state["path"])

1334 self._shallow_from_dict(state)

1335

1336

1337class _WildcardLoad(_AbstractLoad):

1338 """represent a standalone '*' load operation"""

1339

1340 __slots__ = ("strategy", "path", "local_opts")

1341

1342 _traverse_internals = [

1343 ("strategy", visitors.ExtendedInternalTraversal.dp_plain_obj),

1344 ("path", visitors.ExtendedInternalTraversal.dp_plain_obj),

1345 (

1346 "local_opts",

1347 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1348 ),

1349 ]

1350 cache_key_traversal: _CacheKeyTraversalType = None

1351

1352 strategy: Optional[Tuple[Any, ...]]

1353 local_opts: _OptsType

1354 path: Union[Tuple[()], Tuple[str]]

1355 propagate_to_loaders = False

1356

1357 def __init__(self) -> None:

1358 self.path = ()

1359 self.strategy = None

1360 self.local_opts = util.EMPTY_DICT

1361

1362 def _clone_for_bind_strategy(

1363 self,

1364 attrs,

1365 strategy,

1366 wildcard_key,

1367 opts=None,

1368 attr_group=None,

1369 propagate_to_loaders=True,

1370 reconcile_to_other=None,

1371 extra_criteria=None,

1372 ):

1373 assert attrs is not None

1374 attr = attrs[0]

1375 assert (

1376 wildcard_key

1377 and isinstance(attr, str)

1378 and attr in (_WILDCARD_TOKEN, _DEFAULT_TOKEN)

1379 )

1380

1381 attr = f"{wildcard_key}:{attr}"

1382

1383 self.strategy = strategy

1384 self.path = (attr,)

1385 if opts:

1386 self.local_opts = util.immutabledict(opts)

1387

1388 assert extra_criteria is None

1389

1390 def options(self, *opts: _AbstractLoad) -> Self:

1391 raise NotImplementedError("Star option does not support sub-options")

1392

1393 def _apply_to_parent(self, parent: Load) -> None:

1394 """apply this :class:`_orm._WildcardLoad` object as a sub-option of

1395 a :class:`_orm.Load` object.

1396

1397 This method is used by the :meth:`_orm.Load.options` method. Note

1398 that :class:`_orm.WildcardLoad` itself can't have sub-options, but

1399 it may be used as the sub-option of a :class:`_orm.Load` object.

1400

1401 """

1402 assert self.path

1403 attr = self.path[0]

1404 if attr.endswith(_DEFAULT_TOKEN):

1405 attr = f"{attr.split(':')[0]}:{_WILDCARD_TOKEN}"

1406

1407 effective_path = cast(AbstractEntityRegistry, parent.path).token(attr)

1408

1409 assert effective_path.is_token

1410

1411 loader = _TokenStrategyLoad.create(

1412 effective_path,

1413 None,

1414 self.strategy,

1415 None,

1416 self.local_opts,

1417 self.propagate_to_loaders,

1418 )

1419

1420 parent.context += (loader,)

1421

1422 def _process(self, compile_state, mapper_entities, raiseerr):

1423 is_refresh = compile_state.compile_options._for_refresh_state

1424

1425 if is_refresh and not self.propagate_to_loaders:

1426 return

1427

1428 entities = [ent.entity_zero for ent in mapper_entities]

1429 current_path = compile_state.current_path

1430

1431 start_path: _PathRepresentation = self.path

1432

1433 if current_path:

1434 # TODO: no cases in test suite where we actually get

1435 # None back here

1436 new_path = self._chop_path(start_path, current_path)

1437 if new_path is None:

1438 return

1439

1440 # chop_path does not actually "chop" a wildcard token path,

1441 # just returns it

1442 assert new_path == start_path

1443

1444 # start_path is a single-token tuple

1445 assert start_path and len(start_path) == 1

1446

1447 token = start_path[0]

1448 assert isinstance(token, str)

1449 entity = self._find_entity_basestring(entities, token, raiseerr)

1450

1451 if not entity:

1452 return

1453

1454 path_element = entity

1455

1456 # transfer our entity-less state into a Load() object

1457 # with a real entity path. Start with the lead entity

1458 # we just located, then go through the rest of our path

1459 # tokens and populate into the Load().

1460

1461 assert isinstance(token, str)

1462 loader = _TokenStrategyLoad.create(

1463 path_element._path_registry,

1464 token,

1465 self.strategy,

1466 None,

1467 self.local_opts,

1468 self.propagate_to_loaders,

1469 raiseerr=raiseerr,

1470 )

1471 if not loader:

1472 return

1473

1474 assert loader.path.is_token

1475

1476 # don't pass a reconciled lead entity here

1477 loader.process_compile_state(

1478 self, compile_state, mapper_entities, None, raiseerr

1479 )

1480

1481 return loader

1482

1483 def _find_entity_basestring(

1484 self,

1485 entities: Iterable[_InternalEntityType[Any]],

1486 token: str,

1487 raiseerr: bool,

1488 ) -> Optional[_InternalEntityType[Any]]:

1489 if token.endswith(f":{_WILDCARD_TOKEN}"):

1490 if len(list(entities)) != 1:

1491 if raiseerr:

1492 raise sa_exc.ArgumentError(

1493 "Can't apply wildcard ('*') or load_only() "

1494 f"loader option to multiple entities "

1495 f"{', '.join(str(ent) for ent in entities)}. Specify "

1496 "loader options for each entity individually, such as "

1497 f"""{

1498 ", ".join(

1499 f"Load({ent}).some_option('*')"

1500 for ent in entities

1501 )

1502 }."""

1503 )

1504 elif token.endswith(_DEFAULT_TOKEN):

1505 raiseerr = False

1506

1507 for ent in entities:

1508 # return only the first _MapperEntity when searching

1509 # based on string prop name. Ideally object

1510 # attributes are used to specify more exactly.

1511 return ent

1512 else:

1513 if raiseerr:

1514 raise sa_exc.ArgumentError(

1515 "Query has only expression-based entities - "

1516 f'can\'t find property named "{token}".'

1517 )

1518 else:

1519 return None

1520

1521 def __getstate__(self) -> Dict[str, Any]:

1522 d = self._shallow_to_dict()

1523 return d

1524

1525 def __setstate__(self, state: Dict[str, Any]) -> None:

1526 self._shallow_from_dict(state)

1527

1528

1529class _LoadElement(

1530 cache_key.HasCacheKey, traversals.HasShallowCopy, visitors.Traversible

1531):

1532 """represents strategy information to select for a LoaderStrategy

1533 and pass options to it.

1534

1535 :class:`._LoadElement` objects provide the inner datastructure

1536 stored by a :class:`_orm.Load` object and are also the object passed

1537 to methods like :meth:`.LoaderStrategy.setup_query`.

1538

1539 .. versionadded:: 2.0

1540

1541 """

1542

1543 __slots__ = (

1544 "path",

1545 "strategy",

1546 "propagate_to_loaders",

1547 "local_opts",

1548 "_extra_criteria",

1549 "_reconcile_to_other",

1550 )

1551 __visit_name__ = "load_element"

1552

1553 _traverse_internals = [

1554 ("path", visitors.ExtendedInternalTraversal.dp_has_cache_key),

1555 ("strategy", visitors.ExtendedInternalTraversal.dp_plain_obj),

1556 (

1557 "local_opts",

1558 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1559 ),

1560 ("_extra_criteria", visitors.InternalTraversal.dp_clauseelement_list),

1561 ("propagate_to_loaders", visitors.InternalTraversal.dp_plain_obj),

1562 ("_reconcile_to_other", visitors.InternalTraversal.dp_plain_obj),

1563 ]

1564 _cache_key_traversal = None

1565

1566 _extra_criteria: Tuple[Any, ...]

1567

1568 _reconcile_to_other: Optional[bool]

1569 strategy: Optional[_StrategyKey]

1570 path: PathRegistry

1571 propagate_to_loaders: bool

1572

1573 local_opts: util.immutabledict[str, Any]

1574

1575 is_token_strategy: bool

1576 is_class_strategy: bool

1577

1578 def __hash__(self) -> int:

1579 return id(self)

1580

1581 def __eq__(self, other):

1582 return traversals.compare(self, other)

1583

1584 @property

1585 def is_opts_only(self) -> bool:

1586 return bool(self.local_opts and self.strategy is None)

1587

1588 def _clone(self, **kw: Any) -> Self:

1589 cls = self.__class__

1590 s = cls.__new__(cls)

1591

1592 self._shallow_copy_to(s)

1593 return s

1594

1595 def _update_opts(self, **kw: Any) -> _LoadElement:

1596 new = self._clone()

1597 new.local_opts = new.local_opts.union(kw)

1598 return new

1599

1600 def __getstate__(self) -> Dict[str, Any]:

1601 d = self._shallow_to_dict()

1602 d["path"] = self.path.serialize()

1603 return d

1604

1605 def __setstate__(self, state: Dict[str, Any]) -> None:

1606 state["path"] = PathRegistry.deserialize(state["path"])

1607 self._shallow_from_dict(state)

1608

1609 def _raise_for_no_match(self, parent_loader, mapper_entities):

1610 path = parent_loader.path

1611

1612 found_entities = False

1613 for ent in mapper_entities:

1614 ezero = ent.entity_zero

1615 if ezero:

1616 found_entities = True

1617 break

1618

1619 if not found_entities:

1620 raise sa_exc.ArgumentError(

1621 "Query has only expression-based entities; "

1622 f"attribute loader options for {path[0]} can't "

1623 "be applied here."

1624 )

1625 else:

1626 raise sa_exc.ArgumentError(

1627 f"Mapped class {path[0]} does not apply to any of the "

1628 f"root entities in this query, e.g. "

1629 f"""{

1630 ", ".join(

1631 str(x.entity_zero)

1632 for x in mapper_entities if x.entity_zero

1633 )}. Please """

1634 "specify the full path "

1635 "from one of the root entities to the target "

1636 "attribute. "

1637 )

1638

1639 def _adjust_effective_path_for_current_path(

1640 self, effective_path: PathRegistry, current_path: PathRegistry

1641 ) -> Optional[PathRegistry]:

1642 """receives the 'current_path' entry from an :class:`.ORMCompileState`

1643 instance, which is set during lazy loads and secondary loader strategy

1644 loads, and adjusts the given path to be relative to the

1645 current_path.

1646

1647 E.g. given a loader path and current path:

1648

1649 .. sourcecode:: text

1650

1651 lp: User -> orders -> Order -> items -> Item -> keywords -> Keyword

1652

1653 cp: User -> orders -> Order -> items

1654

1655 The adjusted path would be:

1656

1657 .. sourcecode:: text

1658

1659 Item -> keywords -> Keyword

1660

1661

1662 """

1663 chopped_start_path = Load._chop_path(

1664 effective_path.natural_path, current_path

1665 )

1666 if not chopped_start_path:

1667 return None

1668

1669 tokens_removed_from_start_path = len(effective_path) - len(

1670 chopped_start_path

1671 )

1672

1673 loader_lead_path_element = self.path[tokens_removed_from_start_path]

1674

1675 effective_path = PathRegistry.coerce(

1676 (loader_lead_path_element,) + chopped_start_path[1:]

1677 )

1678

1679 return effective_path

1680

1681 def _init_path(

1682 self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

1683 ):

1684 """Apply ORM attributes and/or wildcard to an existing path, producing

1685 a new path.

1686

1687 This method is used within the :meth:`.create` method to initialize

1688 a :class:`._LoadElement` object.

1689

1690 """

1691 raise NotImplementedError()

1692

1693 def _prepare_for_compile_state(

1694 self,

1695 parent_loader,

1696 compile_state,

1697 mapper_entities,

1698 reconciled_lead_entity,

1699 raiseerr,

1700 ):

1701 """implemented by subclasses."""

1702 raise NotImplementedError()

1703

1704 def process_compile_state(

1705 self,

1706 parent_loader,

1707 compile_state,

1708 mapper_entities,

1709 reconciled_lead_entity,

1710 raiseerr,

1711 ):

1712 """populate ORMCompileState.attributes with loader state for this

1713 _LoadElement.

1714

1715 """

1716 keys = self._prepare_for_compile_state(

1717 parent_loader,

1718 compile_state,

1719 mapper_entities,

1720 reconciled_lead_entity,

1721 raiseerr,

1722 )

1723 for key in keys:

1724 if key in compile_state.attributes:

1725 compile_state.attributes[key] = _LoadElement._reconcile(

1726 self, compile_state.attributes[key]

1727 )

1728 else:

1729 compile_state.attributes[key] = self

1730

1731 @classmethod

1732 def create(

1733 cls,

1734 path: PathRegistry,

1735 attr: Union[_AttrType, _StrPathToken, None],

1736 strategy: Optional[_StrategyKey],

1737 wildcard_key: Optional[_WildcardKeyType],

1738 local_opts: Optional[_OptsType],

1739 propagate_to_loaders: bool,

1740 raiseerr: bool = True,

1741 attr_group: Optional[_AttrGroupType] = None,

1742 reconcile_to_other: Optional[bool] = None,

1743 extra_criteria: Optional[Tuple[Any, ...]] = None,

1744 ) -> _LoadElement:

1745 """Create a new :class:`._LoadElement` object."""

1746

1747 opt = cls.__new__(cls)

1748 opt.path = path

1749 opt.strategy = strategy

1750 opt.propagate_to_loaders = propagate_to_loaders

1751 opt.local_opts = (

1752 util.immutabledict(local_opts) if local_opts else util.EMPTY_DICT

1753 )

1754 opt._extra_criteria = ()

1755

1756 if reconcile_to_other is not None:

1757 opt._reconcile_to_other = reconcile_to_other

1758 elif strategy is None and not local_opts:

1759 opt._reconcile_to_other = True

1760 else:

1761 opt._reconcile_to_other = None

1762

1763 path = opt._init_path(

1764 path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

1765 )

1766

1767 if not path:

1768 return None # type: ignore

1769

1770 assert opt.is_token_strategy == path.is_token

1771

1772 opt.path = path

1773 return opt

1774

1775 def __init__(self) -> None:

1776 raise NotImplementedError()

1777

1778 def _recurse(self) -> _LoadElement:

1779 cloned = self._clone()

1780 cloned.path = PathRegistry.coerce(self.path[:] + self.path[-2:])

1781

1782 return cloned

1783

1784 def _prepend_path_from(self, parent: Load) -> _LoadElement:

1785 """adjust the path of this :class:`._LoadElement` to be

1786 a subpath of that of the given parent :class:`_orm.Load` object's

1787 path.

1788

1789 This is used by the :meth:`_orm.Load._apply_to_parent` method,

1790 which is in turn part of the :meth:`_orm.Load.options` method.

1791

1792 """

1793

1794 if not any(

1795 orm_util._entity_corresponds_to_use_path_impl(

1796 elem,

1797 self.path.odd_element(0),

1798 )

1799 for elem in (parent.path.odd_element(-1),)

1800 + parent.additional_source_entities

1801 ):

1802 raise sa_exc.ArgumentError(

1803 f'Attribute "{self.path[1]}" does not link '

1804 f'from element "{parent.path[-1]}".'

1805 )

1806

1807 return self._prepend_path(parent.path)

1808

1809 def _prepend_path(self, path: PathRegistry) -> Self:

1810 cloned = self._clone()

1811

1812 assert cloned.strategy == self.strategy

1813 assert cloned.local_opts == self.local_opts

1814 assert cloned.is_class_strategy == self.is_class_strategy

1815

1816 cloned.path = PathRegistry.coerce(path[0:-1] + cloned.path[:])

1817

1818 return cloned

1819

1820 @staticmethod

1821 def _reconcile(

1822 replacement: _LoadElement, existing: _LoadElement

1823 ) -> _LoadElement:

1824 """define behavior for when two Load objects are to be put into

1825 the context.attributes under the same key.

1826

1827 :param replacement: ``_LoadElement`` that seeks to replace the

1828 existing one

1829

1830 :param existing: ``_LoadElement`` that is already present.

1831

1832 """

1833 # mapper inheritance loading requires fine-grained "block other

1834 # options" / "allow these options to be overridden" behaviors

1835 # see test_poly_loading.py

1836

1837 if replacement._reconcile_to_other:

1838 return existing

1839 elif replacement._reconcile_to_other is False:

1840 return replacement

1841 elif existing._reconcile_to_other:

1842 return replacement

1843 elif existing._reconcile_to_other is False:

1844 return existing

1845

1846 if existing is replacement:

1847 return replacement

1848 elif (

1849 existing.strategy == replacement.strategy

1850 and existing.local_opts == replacement.local_opts

1851 ):

1852 return replacement

1853 elif replacement.is_opts_only:

1854 existing = existing._clone()

1855 existing.local_opts = existing.local_opts.union(

1856 replacement.local_opts

1857 )

1858 existing._extra_criteria += replacement._extra_criteria

1859 return existing

1860 elif existing.is_opts_only:

1861 replacement = replacement._clone()

1862 replacement.local_opts = replacement.local_opts.union(

1863 existing.local_opts

1864 )

1865 replacement._extra_criteria += existing._extra_criteria

1866 return replacement

1867 elif replacement.path.is_token:

1868 # use 'last one wins' logic for wildcard options. this is also

1869 # kind of inconsistent vs. options that are specific paths which

1870 # will raise as below

1871 return replacement

1872

1873 raise sa_exc.InvalidRequestError(

1874 f"Loader strategies for {replacement.path} conflict"

1875 )

1876

1877

1878class _AttributeStrategyLoad(_LoadElement):

1879 """Loader strategies against specific relationship or column paths.

1880

1881 e.g.::

1882

1883 joinedload(User.addresses)

1884 defer(Order.name)

1885 selectinload(User.orders).lazyload(Order.items)

1886

1887 """

1888

1889 __slots__ = ("_of_type", "_path_with_polymorphic_path")

1890

1891 __visit_name__ = "attribute_strategy_load_element"

1892

1893 _traverse_internals = _LoadElement._traverse_internals + [

1894 ("_of_type", visitors.ExtendedInternalTraversal.dp_multi),

1895 (

1896 "_path_with_polymorphic_path",

1897 visitors.ExtendedInternalTraversal.dp_has_cache_key,

1898 ),

1899 ]

1900

1901 _of_type: Union[Mapper[Any], AliasedInsp[Any], None]

1902 _path_with_polymorphic_path: Optional[PathRegistry]

1903

1904 is_class_strategy = False

1905 is_token_strategy = False

1906

1907 def _init_path(

1908 self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

1909 ):

1910 assert attr is not None

1911 self._of_type = None

1912 self._path_with_polymorphic_path = None

1913 insp, _, prop = _parse_attr_argument(attr)

1914

1915 if insp.is_property:

1916 # direct property can be sent from internal strategy logic

1917 # that sets up specific loaders, such as

1918 # emit_lazyload->_lazyload_reverse

1919 # prop = found_property = attr

1920 prop = attr

1921 path = path[prop]

1922

1923 if path.has_entity:

1924 path = path.entity_path

1925 return path

1926

1927 elif not insp.is_attribute:

1928 # should not reach here;

1929 assert False

1930

1931 # here we assume we have user-passed InstrumentedAttribute

1932 if not orm_util._entity_corresponds_to_use_path_impl(

1933 path[-1], attr.parent

1934 ):

1935 if raiseerr:

1936 if attr_group and attr is not attr_group[0]:

1937 raise sa_exc.ArgumentError(

1938 "Can't apply wildcard ('*') or load_only() "

1939 "loader option to multiple entities in the "

1940 "same option. Use separate options per entity."

1941 )

1942 else:

1943 _raise_for_does_not_link(path, str(attr), attr.parent)

1944 else:

1945 return None

1946

1947 # note the essential logic of this attribute was very different in

1948 # 1.4, where there were caching failures in e.g.

1949 # test_relationship_criteria.py::RelationshipCriteriaTest::

1950 # test_selectinload_nested_criteria[True] if an existing

1951 # "_extra_criteria" on a Load object were replaced with that coming

1952 # from an attribute. This appears to have been an artifact of how

1953 # _UnboundLoad / Load interacted together, which was opaque and

1954 # poorly defined.

1955 if extra_criteria:

1956 assert not attr._extra_criteria

1957 self._extra_criteria = extra_criteria

1958 else:

1959 self._extra_criteria = attr._extra_criteria

1960

1961 if getattr(attr, "_of_type", None):

1962 ac = attr._of_type

1963 ext_info = inspect(ac)

1964 self._of_type = ext_info

1965

1966 self._path_with_polymorphic_path = path.entity_path[prop]

1967

1968 path = path[prop][ext_info]

1969

1970 else:

1971 path = path[prop]

1972

1973 if path.has_entity:

1974 path = path.entity_path

1975

1976 return path

1977

1978 def _prepend_path(self, path: PathRegistry) -> Self:

1979 """Override to also prepend the path for _path_with_polymorphic_path.

1980

1981 When using .options() to chain loader options with of_type(), this

1982 ensures that the polymorphic path information is correctly updated

1983 to include the parent path. Fixes issue #13202.

1984 """

1985 cloned = super()._prepend_path(path)

1986

1987 # Also prepend the parent path to _path_with_polymorphic_path if

1988 # present

1989 if self._path_with_polymorphic_path is not None:

1990 cloned._path_with_polymorphic_path = PathRegistry.coerce(

1991 path[0:-1] + self._path_with_polymorphic_path[:]

1992 )

1993

1994 return cloned

1995

1996 def _generate_extra_criteria(self, context):

1997 """Apply the current bound parameters in a QueryContext to the

1998 immediate "extra_criteria" stored with this Load object.

1999

2000 Load objects are typically pulled from the cached version of

2001 the statement from a QueryContext. The statement currently being

2002 executed will have new values (and keys) for bound parameters in the

2003 extra criteria which need to be applied by loader strategies when

2004 they handle this criteria for a result set.

2005

2006 """

2007

2008 assert (

2009 self._extra_criteria

2010 ), "this should only be called if _extra_criteria is present"

2011

2012 orig_query = context.compile_state.select_statement

2013 current_query = context.query

2014

2015 # NOTE: while it seems like we should not do the "apply" operation

2016 # here if orig_query is current_query, skipping it in the "optimized"

2017 # case causes the query to be different from a cache key perspective,

2018 # because we are creating a copy of the criteria which is no longer

2019 # the same identity of the _extra_criteria in the loader option

2020 # itself. cache key logic produces a different key for

2021 # (A, copy_of_A) vs. (A, A), because in the latter case it shortens

2022 # the second part of the key to just indicate on identity.

2023

2024 # if orig_query is current_query:

2025 # not cached yet. just do the and_()

2026 # return and_(*self._extra_criteria)

2027

2028 k1 = orig_query._generate_cache_key()

2029 k2 = current_query._generate_cache_key()

2030

2031 return k2._apply_params_to_element(k1, and_(*self._extra_criteria))

2032

2033 def _set_of_type_info(self, context, current_path):

2034 assert self._path_with_polymorphic_path

2035

2036 pwpi = self._of_type

2037 assert pwpi

2038 if not pwpi.is_aliased_class:

2039 pwpi = inspect(

2040 orm_util.AliasedInsp._with_polymorphic_factory(

2041 pwpi.mapper.base_mapper,

2042 (pwpi.mapper,),

2043 aliased=True,

2044 _use_mapper_path=True,

2045 )

2046 )

2047 start_path = self._path_with_polymorphic_path

2048 if current_path:

2049 new_path = self._adjust_effective_path_for_current_path(

2050 start_path, current_path

2051 )

2052 if new_path is None:

2053 return

2054 start_path = new_path

2055

2056 key = ("path_with_polymorphic", start_path.natural_path)

2057 if key in context:

2058 existing_aliased_insp = context[key]

2059 this_aliased_insp = pwpi

2060 new_aliased_insp = existing_aliased_insp._merge_with(

2061 this_aliased_insp

2062 )

2063 context[key] = new_aliased_insp

2064 else:

2065 context[key] = pwpi

2066

2067 def _prepare_for_compile_state(

2068 self,

2069 parent_loader,

2070 compile_state,

2071 mapper_entities,

2072 reconciled_lead_entity,

2073 raiseerr,

2074 ):

2075 # _AttributeStrategyLoad

2076

2077 current_path = compile_state.current_path

2078 is_refresh = compile_state.compile_options._for_refresh_state

2079 assert not self.path.is_token

2080

2081 if is_refresh and not self.propagate_to_loaders:

2082 return []

2083

2084 if self._of_type:

2085 # apply additional with_polymorphic alias that may have been

2086 # generated. this has to happen even if this is a defaultload

2087 self._set_of_type_info(compile_state.attributes, current_path)

2088

2089 # omit setting loader attributes for a "defaultload" type of option

2090 if not self.strategy and not self.local_opts:

2091 return []

2092

2093 if raiseerr and not reconciled_lead_entity:

2094 self._raise_for_no_match(parent_loader, mapper_entities)

2095

2096 if self.path.has_entity:

2097 effective_path = self.path.parent

2098 else:

2099 effective_path = self.path

2100

2101 if current_path:

2102 assert effective_path is not None

2103 effective_path = self._adjust_effective_path_for_current_path(

2104 effective_path, current_path

2105 )

2106 if effective_path is None:

2107 return []

2108

2109 return [("loader", cast(PathRegistry, effective_path).natural_path)]

2110

2111 def __getstate__(self):

2112 d = super().__getstate__()

2113

2114 # can't pickle this. See

2115 # test_pickled.py -> test_lazyload_extra_criteria_not_supported

2116 # where we should be emitting a warning for the usual case where this

2117 # would be non-None

2118 d["_extra_criteria"] = ()

2119

2120 if self._path_with_polymorphic_path:

2121 d["_path_with_polymorphic_path"] = (

2122 self._path_with_polymorphic_path.serialize()

2123 )

2124

2125 if self._of_type:

2126 if self._of_type.is_aliased_class:

2127 d["_of_type"] = None

2128 elif self._of_type.is_mapper:

2129 d["_of_type"] = self._of_type.class_

2130 else:

2131 assert False, "unexpected object for _of_type"

2132

2133 return d

2134

2135 def __setstate__(self, state):

2136 super().__setstate__(state)

2137

2138 if state.get("_path_with_polymorphic_path", None):

2139 self._path_with_polymorphic_path = PathRegistry.deserialize(

2140 state["_path_with_polymorphic_path"]

2141 )

2142 else:

2143 self._path_with_polymorphic_path = None

2144

2145 if state.get("_of_type", None):

2146 self._of_type = inspect(state["_of_type"])

2147 else:

2148 self._of_type = None

2149

2150

2151class _TokenStrategyLoad(_LoadElement):

2152 """Loader strategies against wildcard attributes

2153

2154 e.g.::

2155

2156 raiseload("*")

2157 Load(User).lazyload("*")

2158 defer("*")

2159 load_only(User.name, User.email) # will create a defer('*')

2160 joinedload(User.addresses).raiseload("*")

2161

2162 """

2163

2164 __visit_name__ = "token_strategy_load_element"

2165

2166 inherit_cache = True

2167 is_class_strategy = False

2168 is_token_strategy = True

2169

2170 def _init_path(

2171 self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

2172 ):

2173 # assert isinstance(attr, str) or attr is None

2174 if attr is not None:

2175 # the only strings accepted here are the wildcard and default

2176 # tokens, either bare or already prefixed with a wildcard key.

2177 # anything else is a leftover from the string based loader

2178 # option API removed in 2.0 and gets the same error as any

2179 # other string. note that testing only for a trailing "*",

2180 # as was formerly the case, lets a name like "addresses.*"

2181 # through to build a loader path that matches nothing

2182 if attr in _ACCEPTED_TOKENS:

2183 if wildcard_key:

2184 attr = f"{wildcard_key}:{attr}"

2185

2186 path = path.token(attr)

2187 return path

2188 else:

2189 raise sa_exc.ArgumentError(

2190 "Strings are not accepted for attribute names in loader "

2191 "options; please use class-bound attributes directly."

2192 )

2193 return path

2194

2195 def _prepare_for_compile_state(

2196 self,

2197 parent_loader,

2198 compile_state,

2199 mapper_entities,

2200 reconciled_lead_entity,

2201 raiseerr,

2202 ):

2203 # _TokenStrategyLoad

2204

2205 current_path = compile_state.current_path

2206 is_refresh = compile_state.compile_options._for_refresh_state

2207

2208 assert self.path.is_token

2209

2210 if is_refresh and not self.propagate_to_loaders:

2211 return []

2212

2213 # omit setting attributes for a "defaultload" type of option

2214 if not self.strategy and not self.local_opts:

2215 return []

2216

2217 effective_path = self.path

2218 if reconciled_lead_entity:

2219 effective_path = PathRegistry.coerce(

2220 (reconciled_lead_entity,) + effective_path.path[1:]

2221 )

2222

2223 if current_path:

2224 new_effective_path = self._adjust_effective_path_for_current_path(

2225 effective_path, current_path

2226 )

2227 if new_effective_path is None:

2228 return []

2229 effective_path = new_effective_path

2230

2231 # for a wildcard token, expand out the path we set

2232 # to encompass everything from the query entity on

2233 # forward. not clear if this is necessary when current_path

2234 # is set.

2235

2236 return [

2237 ("loader", natural_path)

2238 for natural_path in (

2239 cast(

2240 TokenRegistry, effective_path

2241 )._generate_natural_for_superclasses()

2242 )

2243 ]

2244

2245

2246class _ClassStrategyLoad(_LoadElement):

2247 """Loader strategies that deals with a class as a target, not

2248 an attribute path

2249

2250 e.g.::

2251

2252 q = s.query(Person).options(

2253 selectin_polymorphic(Person, [Engineer, Manager])

2254 )

2255

2256 """

2257

2258 inherit_cache = True

2259 is_class_strategy = True

2260 is_token_strategy = False

2261

2262 __visit_name__ = "class_strategy_load_element"

2263

2264 def _init_path(

2265 self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

2266 ):

2267 return path

2268

2269 def _prepare_for_compile_state(

2270 self,

2271 parent_loader,

2272 compile_state,

2273 mapper_entities,

2274 reconciled_lead_entity,

2275 raiseerr,

2276 ):

2277 # _ClassStrategyLoad

2278

2279 current_path = compile_state.current_path

2280 is_refresh = compile_state.compile_options._for_refresh_state

2281

2282 if is_refresh and not self.propagate_to_loaders:

2283 return []

2284

2285 # omit setting attributes for a "defaultload" type of option

2286 if not self.strategy and not self.local_opts:

2287 return []

2288

2289 effective_path = self.path

2290

2291 if current_path:

2292 new_effective_path = self._adjust_effective_path_for_current_path(

2293 effective_path, current_path

2294 )

2295 if new_effective_path is None:

2296 return []

2297 effective_path = new_effective_path

2298

2299 return [("loader", effective_path.natural_path)]

2300

2301

2302def _generate_from_keys(

2303 meth: Callable[..., _AbstractLoad],

2304 keys: Tuple[_AttrType, ...],

2305 chained: bool,

2306 kw: Any,

2307) -> _AbstractLoad:

2308 lead_element: Optional[_AbstractLoad] = None

2309

2310 attr: Any

2311 for is_default, _keys in (True, keys[0:-1]), (False, keys[-1:]):

2312 for attr in _keys:

2313 if isinstance(attr, str):

2314 if attr.startswith("." + _WILDCARD_TOKEN):

2315 util.warn_deprecated(

2316 "The undocumented `.{WILDCARD}` format is "

2317 "deprecated "

2318 "and will be removed in a future version as "

2319 "it is "

2320 "believed to be unused. "

2321 "If you have been using this functionality, "

2322 "please "

2323 "comment on Issue #4390 on the SQLAlchemy project "

2324 "tracker.",

2325 version="1.4",

2326 )

2327 attr = attr[1:]

2328

2329 if attr == _WILDCARD_TOKEN:

2330 if is_default:

2331 raise sa_exc.ArgumentError(

2332 "Wildcard token cannot be followed by "

2333 "another entity",

2334 )

2335

2336 if lead_element is None:

2337 lead_element = _WildcardLoad()

2338

2339 lead_element = meth(lead_element, _DEFAULT_TOKEN, **kw)

2340

2341 else:

2342 raise sa_exc.ArgumentError(

2343 "Strings are not accepted for attribute names in "

2344 "loader options; please use class-bound "

2345 "attributes directly.",

2346 )

2347 else:

2348 if lead_element is None:

2349 _, lead_entity, _ = _parse_attr_argument(attr)

2350 lead_element = Load(lead_entity)

2351

2352 if is_default:

2353 if not chained:

2354 lead_element = lead_element.defaultload(attr)

2355 else:

2356 lead_element = meth(

2357 lead_element, attr, _is_chain=True, **kw

2358 )

2359 else:

2360 lead_element = meth(lead_element, attr, **kw)

2361

2362 assert lead_element

2363 return lead_element

2364

2365

2366def _parse_attr_argument(

2367 attr: _AttrType,

2368) -> Tuple[InspectionAttr, _InternalEntityType[Any], MapperProperty[Any]]:

2369 """parse an attribute or wildcard argument to produce an

2370 :class:`._AbstractLoad` instance.

2371

2372 This is used by the standalone loader strategy functions like

2373 ``joinedload()``, ``defer()``, etc. to produce :class:`_orm.Load` or

2374 :class:`._WildcardLoad` objects.

2375

2376 """

2377 try:

2378 # TODO: need to figure out this None thing being returned by

2379 # inspect(), it should not have None as an option in most cases

2380 # if at all

2381 insp: InspectionAttr = inspect(attr) # type: ignore

2382 except sa_exc.NoInspectionAvailable as err:

2383 raise sa_exc.ArgumentError(

2384 "expected ORM mapped attribute for loader strategy argument"

2385 ) from err

2386

2387 lead_entity: _InternalEntityType[Any]

2388

2389 if insp_is_mapper_property(insp):

2390 lead_entity = insp.parent

2391 prop = insp

2392 elif insp_is_attribute(insp):

2393 lead_entity = insp.parent

2394 prop = insp.prop

2395 else:

2396 raise sa_exc.ArgumentError(

2397 "expected ORM mapped attribute for loader strategy argument"

2398 )

2399

2400 return insp, lead_entity, prop

2401

2402

2403def loader_unbound_fn(fn: _FN) -> _FN:

2404 """decorator that applies docstrings between standalone loader functions

2405 and the loader methods on :class:`._AbstractLoad`.

2406

2407 """

2408 bound_fn = getattr(_AbstractLoad, fn.__name__)

2409 fn_doc = bound_fn.__doc__

2410 bound_fn.__doc__ = f"""Produce a new :class:`_orm.Load` object with the

2411:func:`_orm.{fn.__name__}` option applied.

2412

2413See :func:`_orm.{fn.__name__}` for usage examples.

2414

2415"""

2416

2417 fn.__doc__ = fn_doc

2418 return fn

2419

2420

2421def _expand_column_strategy_attrs(

2422 attrs: Tuple[_AttrType, ...],

2423) -> Tuple[_AttrType, ...]:

2424 return cast(

2425 "Tuple[_AttrType, ...]",

2426 tuple(

2427 a

2428 for attr in attrs

2429 for a in (

2430 cast("QueryableAttribute[Any]", attr)._column_strategy_attrs()

2431 if hasattr(attr, "_column_strategy_attrs")

2432 else (attr,)

2433 )

2434 ),

2435 )

2436

2437

2438# standalone functions follow. docstrings are filled in

2439# by the ``@loader_unbound_fn`` decorator.

2440

2441

2442@loader_unbound_fn

2443def contains_eager(*keys: _AttrType, **kw: Any) -> _AbstractLoad:

2444 return _generate_from_keys(Load.contains_eager, keys, True, kw)

2445

2446

2447@loader_unbound_fn

2448def load_only(*attrs: _AttrType, raiseload: bool = False) -> _AbstractLoad:

2449 # TODO: attrs against different classes. we likely have to

2450 # add some extra state to Load of some kind

2451 attrs = _expand_column_strategy_attrs(attrs)

2452 _, lead_element, _ = _parse_attr_argument(attrs[0])

2453 return Load(lead_element).load_only(*attrs, raiseload=raiseload)

2454

2455

2456@loader_unbound_fn

2457def joinedload(*keys: _AttrType, **kw: Any) -> _AbstractLoad:

2458 return _generate_from_keys(Load.joinedload, keys, False, kw)

2459

2460

2461@loader_unbound_fn

2462def subqueryload(*keys: _AttrType) -> _AbstractLoad:

2463 return _generate_from_keys(Load.subqueryload, keys, False, {})

2464

2465

2466@loader_unbound_fn

2467def selectinload(

2468 *keys: _AttrType, recursion_depth: Optional[int] = None

2469) -> _AbstractLoad:

2470 return _generate_from_keys(

2471 Load.selectinload, keys, False, {"recursion_depth": recursion_depth}

2472 )

2473

2474

2475@loader_unbound_fn

2476def lazyload(*keys: _AttrType) -> _AbstractLoad:

2477 return _generate_from_keys(Load.lazyload, keys, False, {})

2478

2479

2480@loader_unbound_fn

2481def immediateload(

2482 *keys: _AttrType, recursion_depth: Optional[int] = None

2483) -> _AbstractLoad:

2484 return _generate_from_keys(

2485 Load.immediateload, keys, False, {"recursion_depth": recursion_depth}

2486 )

2487

2488

2489@loader_unbound_fn

2490def noload(*keys: _AttrType) -> _AbstractLoad:

2491 return _generate_from_keys(Load.noload, keys, False, {})

2492

2493

2494@loader_unbound_fn

2495def raiseload(*keys: _AttrType, **kw: Any) -> _AbstractLoad:

2496 return _generate_from_keys(Load.raiseload, keys, False, kw)

2497

2498

2499@loader_unbound_fn

2500def defaultload(*keys: _AttrType) -> _AbstractLoad:

2501 return _generate_from_keys(Load.defaultload, keys, False, {})

2502

2503

2504@loader_unbound_fn

2505def defer(

2506 key: _AttrType, *addl_attrs: _AttrType, raiseload: bool = False

2507) -> _AbstractLoad:

2508 if addl_attrs:

2509 util.warn_deprecated(

2510 "The *addl_attrs on orm.defer is deprecated. Please use "

2511 "method chaining in conjunction with defaultload() to "

2512 "indicate a path.",

2513 version="1.3",

2514 )

2515

2516 if raiseload:

2517 kw = {"raiseload": raiseload}

2518 else:

2519 kw = {}

2520

2521 return _generate_from_keys(Load.defer, (key,) + addl_attrs, False, kw)

2522

2523

2524@loader_unbound_fn

2525def undefer(key: _AttrType, *addl_attrs: _AttrType) -> _AbstractLoad:

2526 if addl_attrs:

2527 util.warn_deprecated(

2528 "The *addl_attrs on orm.undefer is deprecated. Please use "

2529 "method chaining in conjunction with defaultload() to "

2530 "indicate a path.",

2531 version="1.3",

2532 )

2533 return _generate_from_keys(Load.undefer, (key,) + addl_attrs, False, {})

2534

2535

2536@loader_unbound_fn

2537def undefer_group(name: str) -> _AbstractLoad:

2538 element = _WildcardLoad()

2539 return element.undefer_group(name)

2540

2541

2542@loader_unbound_fn

2543def with_expression(

2544 key: _AttrType, expression: _ColumnExpressionArgument[Any]

2545) -> _AbstractLoad:

2546 return _generate_from_keys(

2547 Load.with_expression, (key,), False, {"expression": expression}

2548 )

2549

2550

2551@loader_unbound_fn

2552def selectin_polymorphic(

2553 base_cls: _EntityType[Any], classes: Iterable[Type[Any]]

2554) -> _AbstractLoad:

2555 ul = Load(base_cls)

2556 return ul.selectin_polymorphic(classes)

2557

2558

2559def _raise_for_does_not_link(path, attrname, parent_entity):

2560 if len(path) > 1:

2561 path_is_of_type = path[-1].entity is not path[-2].mapper.class_

2562 if insp_is_aliased_class(parent_entity):

2563 parent_entity_str = str(parent_entity)

2564 else:

2565 parent_entity_str = parent_entity.class_.__name__

2566

2567 raise sa_exc.ArgumentError(

2568 f'ORM mapped entity or attribute "{attrname}" does not '

2569 f'link from relationship "{path[-2]}%s".%s'

2570 % (

2571 f".of_type({path[-1]})" if path_is_of_type else "",

2572 (

2573 " Did you mean to use "

2574 f'"{path[-2]}'

2575 f'.of_type({parent_entity_str})" or "loadopt.options('

2576 f"selectin_polymorphic({path[-2].mapper.class_.__name__}, "

2577 f'[{parent_entity_str}]), ...)" ?'

2578 if not path_is_of_type

2579 and not path[-1].is_aliased_class

2580 and orm_util._entity_corresponds_to(

2581 path.entity, inspect(parent_entity).mapper

2582 )

2583 else ""

2584 ),

2585 )

2586 )

2587 else:

2588 raise sa_exc.ArgumentError(

2589 f'ORM mapped attribute "{attrname}" does not '

2590 f'link mapped class "{path[-1]}"'

2591 )

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

790 statements