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 Final

19from typing import Iterable

20from typing import Literal

21from typing import Optional

22from typing import overload

23from typing import Sequence

24from typing import Tuple

25from typing import Type

26from typing import TypeVar

27from typing import Union

29from . import util as orm_util

30from ._typing import insp_is_aliased_class

31from ._typing import insp_is_attribute

32from ._typing import insp_is_mapper

33from ._typing import insp_is_mapper_property

34from .attributes import QueryableAttribute

35from .base import InspectionAttr

36from .interfaces import LoaderOption

37from .path_registry import _AbstractEntityRegistry

38from .path_registry import _DEFAULT_TOKEN

39from .path_registry import _StrPathToken

40from .path_registry import _TokenRegistry

41from .path_registry import _WILDCARD_TOKEN

42from .path_registry import path_is_property

43from .path_registry import PathRegistry

44from .util import _orm_full_deannotate

45from .util import AliasedInsp

46from .. import exc as sa_exc

47from .. import inspect

48from .. import util

49from ..sql import and_

50from ..sql import cache_key

51from ..sql import coercions

52from ..sql import roles

53from ..sql import traversals

54from ..sql import visitors

55from ..sql.base import _generative

56from ..util.typing import Self

58_RELATIONSHIP_TOKEN: Final[Literal["relationship"]] = "relationship"

59_COLUMN_TOKEN: Final[Literal["column"]] = "column"

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

63if typing.TYPE_CHECKING:

64 from ._typing import _EntityType

65 from ._typing import _InternalEntityType

66 from .context import _MapperEntity

67 from .context import _ORMCompileState

68 from .context import QueryContext

69 from .interfaces import _StrategyKey

70 from .interfaces import MapperProperty

71 from .interfaces import ORMOption

72 from .mapper import Mapper

73 from .path_registry import _PathRepresentation

74 from ..sql._typing import _ColumnExpressionArgument

75 from ..sql._typing import _FromClauseArgument

76 from ..sql.cache_key import _CacheKeyTraversalType

77 from ..sql.cache_key import CacheKey

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

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

83_StrategySpec = Dict[str, Any]

84_OptsType = Dict[str, Any]

85_AttrGroupType = Tuple[_AttrType, ...]

88class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):

89 __slots__ = ("propagate_to_loaders",)

91 _is_strategy_option = True

92 propagate_to_loaders: bool

94 def contains_eager(

95 self,

96 attr: _AttrType,

97 alias: Optional[_FromClauseArgument] = None,

98 _is_chain: bool = False,

99 _propagate_to_loaders: bool = False,

100 ) -> Self:

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

102 columns stated manually in the query.

103

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

105 both method-chained and standalone operation.

106

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

108 the desired rows, i.e.::

109

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

111

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

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

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

115

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

117 collection; queries will normally want to use the

118 :ref:`orm_queryguide_populate_existing` execution option assuming the

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

120

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

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

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

124

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

126

127 .. seealso::

128

129 :ref:`loading_toplevel`

130

131 :ref:`contains_eager`

132

133 """

134 if alias is not None:

135 if not isinstance(alias, str):

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

137 else:

138 util.warn_deprecated(

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

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

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

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

143 version="1.4",

144 )

145 coerced_alias = alias

146

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

148 assert isinstance(attr, QueryableAttribute)

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

150 assert ot is not None

151 coerced_alias = ot.selectable

152 else:

153 coerced_alias = None

154

155 cloned = self._set_relationship_strategy(

156 attr,

157 {"lazy": "joined"},

158 propagate_to_loaders=_propagate_to_loaders,

159 opts={"eager_from_alias": coerced_alias},

160 _reconcile_to_other=True if _is_chain else None,

161 )

162 return cloned

163

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

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

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

167 deferred.

168

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

170 both method-chained and standalone operation.

171

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

173 ``fullname`` attributes::

174

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

176

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

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

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

180

181 session.query(User).options(

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

183 )

184

185 For a statement that has multiple entities,

186 the lead entity can be

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

188

189 stmt = (

190 select(User, Address)

191 .join(User.addresses)

192 .options(

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

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

195 )

196 )

197

198 When used together with the

199 :ref:`populate_existing <orm_queryguide_populate_existing>`

200 execution option only the attributes listed will be refreshed.

201

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

203

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

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

206 to prevent unwanted SQL from being emitted.

207

208 .. versionadded:: 2.0

209

210 .. seealso::

211

212 :ref:`orm_queryguide_column_deferral` - in the

213 :ref:`queryguide_toplevel`

214

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

216

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

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

219 to prevent unwanted SQL from being emitted.

220

221 .. versionadded:: 2.0

222

223 """

224 cloned = self._set_column_strategy(

225 _expand_column_strategy_attrs(attrs),

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

227 )

228

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

230 if raiseload:

231 wildcard_strategy["raiseload"] = True

232

233 cloned = cloned._set_column_strategy(

234 ("*",),

235 wildcard_strategy,

236 )

237 return cloned

238

239 def joinedload(

240 self,

241 attr: _AttrType,

242 innerjoin: Optional[bool] = None,

243 ) -> Self:

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

245 eager loading.

246

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

248 both method-chained and standalone operation.

249

250 examples::

251

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

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

254

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

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

257

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

259 # joined-load the keywords collection

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

261

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

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

264

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

266

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

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

269

270 select(A).options(

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

272 )

273

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

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

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

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

278 directly supported.

279

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

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

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

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

284 is an outerjoin::

285

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

287

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

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

290

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

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

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

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

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

296

297 .. note::

298

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

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

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

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

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

304

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

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

307 explicit JOINs with eager loading of collections, use

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

309

310 .. seealso::

311

312 :ref:`loading_toplevel`

313

314 :ref:`joined_eager_loading`

315

316 """ # noqa: E501

317 loader = self._set_relationship_strategy(

318 attr,

319 {"lazy": "joined"},

320 opts=(

321 {"innerjoin": innerjoin}

322 if innerjoin is not None

323 else util.EMPTY_DICT

324 ),

325 )

326 return loader

327

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

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

330 subquery eager loading.

331

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

333 both method-chained and standalone operation.

334

335 examples::

336

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

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

339

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

341 select(Order).options(

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

343 )

344

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

346 # subquery-load the keywords collection

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

348

349 .. seealso::

350

351 :ref:`loading_toplevel`

352

353 :ref:`subquery_eager_loading`

354

355 """

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

357

358 def selectinload(

359 self,

360 attr: _AttrType,

361 recursion_depth: Optional[int] = None,

362 chunksize: Optional[int] = None,

363 ) -> Self:

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

365 SELECT IN eager loading.

366

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

368 both method-chained and standalone operation.

369

370 examples::

371

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

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

374

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

376 select(Order).options(

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

378 )

379

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

381 # selectin-load the keywords collection

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

383

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

385 in conjunction with a self-referential relationship,

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

387 automatically until no items are found.

388

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

390 currently supports only self-referential relationships. There

391 is not yet an option to automatically traverse recursive structures

392 with more than one relationship involved.

393

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

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

396 status for the 2.0 series.

397

398 .. versionadded:: 2.0 added

399 :paramref:`_orm.selectinload.recursion_depth`

400

401 :param chunksize: optional int; when set to a positive non-zero

402 integer, the keys from the IN statement will be chunked relative

403 to the passed parameter

404

405 .. versionadded:: 2.1.0b3

406

407 .. seealso::

408

409 :ref:`loading_toplevel`

410

411 :ref:`selectin_eager_loading`

412

413 """

414 return self._set_relationship_strategy(

415 attr,

416 {"lazy": "selectin"},

417 opts={"recursion_depth": recursion_depth, "chunksize": chunksize},

418 )

419

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

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

422 loading.

423

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

425 both method-chained and standalone operation.

426

427 .. seealso::

428

429 :ref:`loading_toplevel`

430

431 :ref:`lazy_loading`

432

433 """

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

435

436 def immediateload(

437 self,

438 attr: _AttrType,

439 recursion_depth: Optional[int] = None,

440 ) -> Self:

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

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

443

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

445 fire off any additional eager loaders.

446

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

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

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

450

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

452 both method-chained and standalone operation.

453

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

455 in conjunction with a self-referential relationship,

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

457 automatically until no items are found.

458

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

460 currently supports only self-referential relationships. There

461 is not yet an option to automatically traverse recursive structures

462 with more than one relationship involved.

463

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

465 treated as "alpha" status

466

467 .. versionadded:: 2.0 added

468 :paramref:`_orm.immediateload.recursion_depth`

469

470

471 .. seealso::

472

473 :ref:`loading_toplevel`

474

475 :ref:`selectin_eager_loading`

476

477 """

478 loader = self._set_relationship_strategy(

479 attr,

480 {"lazy": "immediate"},

481 opts={"recursion_depth": recursion_depth},

482 )

483 return loader

484

485 @util.deprecated(

486 "2.1",

487 "The :func:`_orm.noload` option is deprecated and will be removed "

488 "in a future release. This option "

489 "produces incorrect results by returning ``None`` for related "

490 "items.",

491 )

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

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

494 unloaded.

495

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

497 producing any loading effect.

498

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

500 only.

501

502 .. seealso::

503

504 :ref:`loading_toplevel`

505

506 """

507

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

509

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

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

512

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

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

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

516 ensure that all relationship attributes that are accessed in a

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

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

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

520

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

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

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

524 :func:`.defer` loader option.

525

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

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

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

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

530

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

532 both method-chained and standalone operation.

533

534 .. seealso::

535

536 :ref:`loading_toplevel`

537

538 :ref:`prevent_lazy_with_raiseload`

539

540 :ref:`orm_queryguide_deferred_raiseload`

541

542 """

543

544 return self._set_relationship_strategy(

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

546 )

547

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

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

550

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

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

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

554 loading will be used.

555

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

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

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

559 element of an element::

560

561 session.query(MyClass).options(

562 defaultload(MyClass.someattribute).joinedload(

563 MyOtherClass.someotherattribute

564 )

565 )

566

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

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

569

570 session.scalars(

571 select(MyClass).options(

572 defaultload(MyClass.someattribute)

573 .defer("some_column")

574 .undefer("some_other_column")

575 )

576 )

577

578 .. seealso::

579

580 :ref:`orm_queryguide_relationship_sub_options`

581

582 :meth:`_orm.Load.options`

583

584 """

585 return self._set_relationship_strategy(attr, None)

586

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

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

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

590

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

592 both method-chained and standalone operation.

593

594 e.g.::

595

596 from sqlalchemy.orm import defer

597

598 session.query(MyClass).options(

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

600 )

601

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

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

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

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

606

607 session.query(MyClass).options(

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

609 )

610

611 Multiple deferral options related to a relationship can be bundled

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

613

614

615 select(MyClass).options(

616 defaultload(MyClass.someattr).options(

617 defer(RelatedClass.some_column),

618 defer(RelatedClass.some_other_column),

619 defer(RelatedClass.another_column),

620 )

621 )

622

623 :param key: Attribute to be deferred.

624

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

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

627 to prevent unwanted SQL from being emitted.

628

629 .. versionadded:: 1.4

630

631 .. seealso::

632

633 :ref:`orm_queryguide_column_deferral` - in the

634 :ref:`queryguide_toplevel`

635

636 :func:`_orm.load_only`

637

638 :func:`_orm.undefer`

639

640 """

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

642 if raiseload:

643 strategy["raiseload"] = True

644 return self._set_column_strategy(

645 _expand_column_strategy_attrs((key,)), strategy

646 )

647

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

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

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

651 as a whole.

652

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

654 :func:`.deferred` attribute.

655

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

657 both method-chained and standalone operation.

658

659 Examples::

660

661 # undefer two columns

662 session.query(MyClass).options(

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

664 )

665

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

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

668

669 # undefer a column on a related object

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

671

672 :param key: Attribute to be undeferred.

673

674 .. seealso::

675

676 :ref:`orm_queryguide_column_deferral` - in the

677 :ref:`queryguide_toplevel`

678

679 :func:`_orm.defer`

680

681 :func:`_orm.undefer_group`

682

683 """ # noqa: E501

684 return self._set_column_strategy(

685 _expand_column_strategy_attrs((key,)),

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

687 )

688

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

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

691 undeferred.

692

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

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

695

696 E.g::

697

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

699

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

701 spelled out using relationship loader options, such as

702 :func:`_orm.defaultload`::

703

704 select(MyClass).options(

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

706 )

707

708 .. seealso::

709

710 :ref:`orm_queryguide_column_deferral` - in the

711 :ref:`queryguide_toplevel`

712

713 :func:`_orm.defer`

714

715 :func:`_orm.undefer`

716

717 """

718 return self._set_column_strategy(

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

720 )

721

722 def with_expression(

723 self,

724 key: _AttrType,

725 expression: _ColumnExpressionArgument[Any],

726 ) -> Self:

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

728 attribute.

729

730 This option is used in conjunction with the

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

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

733

734 E.g.::

735

736 stmt = select(SomeClass).options(

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

738 )

739

740 :param key: Attribute to be populated

741

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

743

744 .. seealso::

745

746 :ref:`orm_queryguide_with_expression` - background and usage

747 examples

748

749 """

750

751 expression = _orm_full_deannotate(

752 coercions.expect(roles.LabeledColumnExprRole, expression)

753 )

754

755 return self._set_column_strategy(

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

757 )

758

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

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

761 specific to a subclass.

762

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

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

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

766

767 .. seealso::

768

769 :ref:`polymorphic_selectin`

770

771 """

772 self = self._set_class_strategy(

773 {"selectinload_polymorphic": True},

774 opts={

775 "entities": tuple(

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

777 )

778 },

779 )

780 return self

781

782 @overload

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

784

785 @overload

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

787

788 def _coerce_strat(

789 self, strategy: Optional[_StrategySpec]

790 ) -> Optional[_StrategyKey]:

791 if strategy is not None:

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

793 else:

794 strategy_key = None

795 return strategy_key

796

797 @_generative

798 def _set_relationship_strategy(

799 self,

800 attr: _AttrType,

801 strategy: Optional[_StrategySpec],

802 propagate_to_loaders: bool = True,

803 opts: Optional[_OptsType] = None,

804 _reconcile_to_other: Optional[bool] = None,

805 ) -> Self:

806 strategy_key = self._coerce_strat(strategy)

807

808 self._clone_for_bind_strategy(

809 (attr,),

810 strategy_key,

811 _RELATIONSHIP_TOKEN,

812 opts=opts,

813 propagate_to_loaders=propagate_to_loaders,

814 reconcile_to_other=_reconcile_to_other,

815 )

816 return self

817

818 @_generative

819 def _set_column_strategy(

820 self,

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

822 strategy: Optional[_StrategySpec],

823 opts: Optional[_OptsType] = None,

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

825 ) -> Self:

826 strategy_key = self._coerce_strat(strategy)

827

828 self._clone_for_bind_strategy(

829 attrs,

830 strategy_key,

831 _COLUMN_TOKEN,

832 opts=opts,

833 attr_group=attrs,

834 extra_criteria=extra_criteria,

835 )

836 return self

837

838 @_generative

839 def _set_generic_strategy(

840 self,

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

842 strategy: _StrategySpec,

843 _reconcile_to_other: Optional[bool] = None,

844 ) -> Self:

845 strategy_key = self._coerce_strat(strategy)

846 self._clone_for_bind_strategy(

847 attrs,

848 strategy_key,

849 None,

850 propagate_to_loaders=True,

851 reconcile_to_other=_reconcile_to_other,

852 )

853 return self

854

855 @_generative

856 def _set_class_strategy(

857 self, strategy: _StrategySpec, opts: _OptsType

858 ) -> Self:

859 strategy_key = self._coerce_strat(strategy)

860

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

862 return self

863

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

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

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

867

868 Implementation is provided by subclasses.

869

870 """

871 raise NotImplementedError()

872

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

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

875 :class:`_orm._AbstractLoad` object.

876

877 Implementation is provided by subclasses.

878

879 """

880 raise NotImplementedError()

881

882 def _clone_for_bind_strategy(

883 self,

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

885 strategy: Optional[_StrategyKey],

886 wildcard_key: Optional[_WildcardKeyType],

887 opts: Optional[_OptsType] = None,

888 attr_group: Optional[_AttrGroupType] = None,

889 propagate_to_loaders: bool = True,

890 reconcile_to_other: Optional[bool] = None,

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

892 ) -> Self:

893 raise NotImplementedError()

894

895 def process_compile_state_replaced_entities(

896 self,

897 compile_state: _ORMCompileState,

898 mapper_entities: Sequence[_MapperEntity],

899 ) -> None:

900 if not compile_state.compile_options._enable_eagerloads:

901 return

902

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

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

905 # for the entities having been replaced with equivalents

906 self._process(

907 compile_state,

908 mapper_entities,

909 not bool(compile_state.current_path),

910 )

911

912 def process_compile_state(self, compile_state: _ORMCompileState) -> None:

913 if not compile_state.compile_options._enable_eagerloads:

914 return

915

916 self._process(

917 compile_state,

918 compile_state._lead_mapper_entities,

919 not bool(compile_state.current_path)

920 and not compile_state.compile_options._for_refresh_state,

921 )

922

923 def _process(

924 self,

925 compile_state: _ORMCompileState,

926 mapper_entities: Sequence[_MapperEntity],

927 raiseerr: bool,

928 ) -> None:

929 """implemented by subclasses"""

930 raise NotImplementedError()

931

932 @classmethod

933 def _chop_path(

934 cls,

935 to_chop: _PathRepresentation,

936 path: PathRegistry,

937 debug: bool = False,

938 ) -> Optional[_PathRepresentation]:

939 i = -1

940

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

942 zip(to_chop, path.natural_path)

943 ):

944 if isinstance(c_token, str):

945 if i == 0 and (

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

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

948 ):

949 return to_chop

950 elif (

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

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

953 ):

954 return None

955

956 if c_token is p_token:

957 continue

958 elif (

959 isinstance(c_token, InspectionAttr)

960 and insp_is_mapper(c_token)

961 and insp_is_mapper(p_token)

962 and c_token.isa(p_token)

963 ):

964 continue

965

966 else:

967 return None

968 return to_chop[i + 1 :]

969

970

971class Load(_AbstractLoad):

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

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

974 order to affect how various mapped attributes are loaded.

975

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

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

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

979 except for in some very specific cases.

980

981 .. seealso::

982

983 :ref:`orm_queryguide_relationship_per_entity_wildcard` - illustrates an

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

985

986 """

987

988 __slots__ = (

989 "path",

990 "context",

991 "additional_source_entities",

992 )

993

994 _traverse_internals = [

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

996 (

997 "context",

998 visitors.InternalTraversal.dp_has_cache_key_list,

999 ),

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

1001 (

1002 "additional_source_entities",

1003 visitors.InternalTraversal.dp_has_cache_key_list,

1004 ),

1005 ]

1006 _cache_key_traversal = None

1007

1008 path: PathRegistry

1009 context: Tuple[_LoadElement, ...]

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

1011

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

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

1014 insp._post_inspect

1015

1016 self.path = insp._path_registry

1017 self.context = ()

1018 self.propagate_to_loaders = False

1019 self.additional_source_entities = ()

1020

1021 def __str__(self) -> str:

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

1023

1024 @classmethod

1025 def _construct_for_existing_path(

1026 cls, path: _AbstractEntityRegistry

1027 ) -> Load:

1028 load = cls.__new__(cls)

1029 load.path = path

1030 load.context = ()

1031 load.propagate_to_loaders = False

1032 load.additional_source_entities = ()

1033 return load

1034

1035 def _adapt_cached_option_to_uncached_option(

1036 self, context: QueryContext, uncached_opt: ORMOption

1037 ) -> ORMOption:

1038 if uncached_opt is self:

1039 return self

1040 return self._adjust_for_extra_criteria(context)

1041

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

1043 cloned = self._clone()

1044 cloned.context = tuple(

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

1046 )

1047 return cloned

1048

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

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

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

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

1053

1054 """

1055

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

1057 # actually have any extra_criteria options, which is the

1058 # common case

1059 for value in self.context:

1060 if value._extra_criteria:

1061 break

1062 else:

1063 return self

1064

1065 replacement_cache_key = context.user_passed_query._generate_cache_key()

1066

1067 if replacement_cache_key is None:

1068 return self

1069

1070 orig_query = context.compile_state.select_statement

1071 orig_cache_key = orig_query._generate_cache_key()

1072 assert orig_cache_key is not None

1073

1074 def process(

1075 opt: _LoadElement,

1076 replacement_cache_key: CacheKey,

1077 orig_cache_key: CacheKey,

1078 ) -> _LoadElement:

1079 cloned_opt = opt._clone()

1080

1081 cloned_opt._extra_criteria = tuple(

1082 replacement_cache_key._apply_params_to_element(

1083 orig_cache_key, crit

1084 )

1085 for crit in cloned_opt._extra_criteria

1086 )

1087

1088 return cloned_opt

1089

1090 cloned = self._clone()

1091 cloned.context = tuple(

1092 (

1093 process(value, replacement_cache_key, orig_cache_key)

1094 if value._extra_criteria

1095 else value

1096 )

1097 for value in self.context

1098 )

1099 return cloned

1100

1101 def _reconcile_query_entities_with_us(self, mapper_entities, raiseerr):

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

1103 entity inside of _LoadElement objects.

1104

1105 """

1106 path = self.path

1107

1108 for ent in mapper_entities:

1109 ezero = ent.entity_zero

1110 if ezero and orm_util._entity_corresponds_to(

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

1112 # safe to pass to _entity_corresponds_to()

1113 ezero,

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

1115 ):

1116 return ezero

1117

1118 return None

1119

1120 def _process(

1121 self,

1122 compile_state: _ORMCompileState,

1123 mapper_entities: Sequence[_MapperEntity],

1124 raiseerr: bool,

1125 ) -> None:

1126 reconciled_lead_entity = self._reconcile_query_entities_with_us(

1127 mapper_entities, raiseerr

1128 )

1129

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

1131 has_current_path = bool(compile_state.compile_options._current_path)

1132

1133 for loader in self.context:

1134 # issue #11292

1135 # historically, propagate_to_loaders was only considered at

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

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

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

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

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

1141 # so we check again

1142 if has_current_path and not loader.propagate_to_loaders:

1143 continue

1144 loader.process_compile_state(

1145 self,

1146 compile_state,

1147 mapper_entities,

1148 reconciled_lead_entity,

1149 raiseerr,

1150 )

1151

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

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

1154 :class:`_orm.Load` object.

1155

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

1157

1158 """

1159 cloned = self._generate()

1160

1161 assert cloned.propagate_to_loaders == self.propagate_to_loaders

1162

1163 if not any(

1164 orm_util._entity_corresponds_to_use_path_impl(

1165 elem, cloned.path.odd_element(0)

1166 )

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

1168 + parent.additional_source_entities

1169 ):

1170 if len(cloned.path) > 1:

1171 attrname = cloned.path[1]

1172 parent_entity = cloned.path[0]

1173 else:

1174 attrname = cloned.path[0]

1175 parent_entity = cloned.path[0]

1176 _raise_for_does_not_link(parent.path, attrname, parent_entity)

1177

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

1179

1180 if self.context:

1181 cloned.context = tuple(

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

1183 )

1184

1185 if cloned.context:

1186 parent.context += cloned.context

1187 parent.additional_source_entities += (

1188 cloned.additional_source_entities

1189 )

1190

1191 @_generative

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

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

1194 :class:`_orm.Load`

1195 object.

1196

1197 E.g.::

1198

1199 query = session.query(Author)

1200 query = query.options(

1201 joinedload(Author.book).options(

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

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

1204 )

1205 )

1206

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

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

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

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 default_token = attr.endswith(_DEFAULT_TOKEN)

2176 if attr.endswith(_WILDCARD_TOKEN) or default_token:

2177 if wildcard_key:

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

2179

2180 path = path.token(attr)

2181 return path

2182 else:

2183 raise sa_exc.ArgumentError(

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

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

2186 )

2187 return path

2188

2189 def _prepare_for_compile_state(

2190 self,

2191 parent_loader,

2192 compile_state,

2193 mapper_entities,

2194 reconciled_lead_entity,

2195 raiseerr,

2196 ):

2197 # _TokenStrategyLoad

2198

2199 current_path = compile_state.current_path

2200 is_refresh = compile_state.compile_options._for_refresh_state

2201

2202 assert self.path.is_token

2203

2204 if is_refresh and not self.propagate_to_loaders:

2205 return []

2206

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

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

2209 return []

2210

2211 effective_path = self.path

2212 if reconciled_lead_entity:

2213 effective_path = PathRegistry.coerce(

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

2215 )

2216

2217 if current_path:

2218 new_effective_path = self._adjust_effective_path_for_current_path(

2219 effective_path, current_path

2220 )

2221 if new_effective_path is None:

2222 return []

2223 effective_path = new_effective_path

2224

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

2226 # to encompass everything from the query entity on

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

2228 # is set.

2229

2230 return [

2231 ("loader", natural_path)

2232 for natural_path in (

2233 cast(

2234 _TokenRegistry, effective_path

2235 )._generate_natural_for_superclasses()

2236 )

2237 ]

2238

2239

2240class _ClassStrategyLoad(_LoadElement):

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

2242 an attribute path

2243

2244 e.g.::

2245

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

2247 selectin_polymorphic(Person, [Engineer, Manager])

2248 )

2249

2250 """

2251

2252 inherit_cache = True

2253 is_class_strategy = True

2254 is_token_strategy = False

2255

2256 __visit_name__ = "class_strategy_load_element"

2257

2258 def _init_path(

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

2260 ):

2261 return path

2262

2263 def _prepare_for_compile_state(

2264 self,

2265 parent_loader,

2266 compile_state,

2267 mapper_entities,

2268 reconciled_lead_entity,

2269 raiseerr,

2270 ):

2271 # _ClassStrategyLoad

2272

2273 current_path = compile_state.current_path

2274 is_refresh = compile_state.compile_options._for_refresh_state

2275

2276 if is_refresh and not self.propagate_to_loaders:

2277 return []

2278

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

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

2281 return []

2282

2283 effective_path = self.path

2284

2285 if current_path:

2286 new_effective_path = self._adjust_effective_path_for_current_path(

2287 effective_path, current_path

2288 )

2289 if new_effective_path is None:

2290 return []

2291 effective_path = new_effective_path

2292

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

2294

2295

2296def _generate_from_keys(

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

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

2299 chained: bool,

2300 kw: Any,

2301) -> _AbstractLoad:

2302 lead_element: Optional[_AbstractLoad] = None

2303

2304 attr: Any

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

2306 for attr in _keys:

2307 if isinstance(attr, str):

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

2309 util.warn_deprecated(

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

2311 "deprecated "

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

2313 "it is "

2314 "believed to be unused. "

2315 "If you have been using this functionality, "

2316 "please "

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

2318 "tracker.",

2319 version="1.4",

2320 )

2321 attr = attr[1:]

2322

2323 if attr == _WILDCARD_TOKEN:

2324 if is_default:

2325 raise sa_exc.ArgumentError(

2326 "Wildcard token cannot be followed by "

2327 "another entity",

2328 )

2329

2330 if lead_element is None:

2331 lead_element = _WildcardLoad()

2332

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

2334

2335 else:

2336 raise sa_exc.ArgumentError(

2337 "Strings are not accepted for attribute names in "

2338 "loader options; please use class-bound "

2339 "attributes directly.",

2340 )

2341 else:

2342 if lead_element is None:

2343 _, lead_entity, _ = _parse_attr_argument(attr)

2344 lead_element = Load(lead_entity)

2345

2346 if is_default:

2347 if not chained:

2348 lead_element = lead_element.defaultload(attr)

2349 else:

2350 lead_element = meth(

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

2352 )

2353 else:

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

2355

2356 assert lead_element

2357 return lead_element

2358

2359

2360def _parse_attr_argument(

2361 attr: _AttrType,

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

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

2364 :class:`._AbstractLoad` instance.

2365

2366 This is used by the standalone loader strategy functions like

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

2368 :class:`._WildcardLoad` objects.

2369

2370 """

2371 try:

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

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

2374 # if at all

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

2376 except sa_exc.NoInspectionAvailable as err:

2377 raise sa_exc.ArgumentError(

2378 "expected ORM mapped attribute for loader strategy argument"

2379 ) from err

2380

2381 lead_entity: _InternalEntityType[Any]

2382

2383 if insp_is_mapper_property(insp):

2384 lead_entity = insp.parent

2385 prop = insp

2386 elif insp_is_attribute(insp):

2387 lead_entity = insp.parent

2388 prop = insp.prop

2389 else:

2390 raise sa_exc.ArgumentError(

2391 "expected ORM mapped attribute for loader strategy argument"

2392 )

2393

2394 return insp, lead_entity, prop

2395

2396

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

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

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

2400

2401 """

2402 bound_fn = getattr(_AbstractLoad, fn.__name__)

2403 fn_doc = bound_fn.__doc__

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

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

2406

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

2408

2409"""

2410

2411 fn.__doc__ = fn_doc

2412 return fn

2413

2414

2415def _expand_column_strategy_attrs(

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

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

2418 return cast(

2419 "Tuple[_AttrType, ...]",

2420 tuple(

2421 a

2422 for attr in attrs

2423 for a in (

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

2425 if hasattr(attr, "_column_strategy_attrs")

2426 else (attr,)

2427 )

2428 ),

2429 )

2430

2431

2432# standalone functions follow. docstrings are filled in

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

2434

2435

2436@loader_unbound_fn

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

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

2439

2440

2441@loader_unbound_fn

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

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

2444 # add some extra state to Load of some kind

2445 attrs = _expand_column_strategy_attrs(attrs)

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

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

2448

2449

2450@loader_unbound_fn

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

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

2453

2454

2455@loader_unbound_fn

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

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

2458

2459

2460@loader_unbound_fn

2461def selectinload(

2462 *keys: _AttrType,

2463 recursion_depth: Optional[int] = None,

2464 chunksize: Optional[int] = None,

2465) -> _AbstractLoad:

2466 return _generate_from_keys(

2467 Load.selectinload,

2468 keys,

2469 False,

2470 {"recursion_depth": recursion_depth, "chunksize": chunksize},

2471 )

2472

2473

2474@loader_unbound_fn

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

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

2477

2478

2479@loader_unbound_fn

2480def immediateload(

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

2482) -> _AbstractLoad:

2483 return _generate_from_keys(

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

2485 )

2486

2487

2488@loader_unbound_fn

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

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

2491

2492

2493@loader_unbound_fn

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

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

2496

2497

2498@loader_unbound_fn

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

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

2501

2502

2503@loader_unbound_fn

2504def defer(key: _AttrType, *, raiseload: bool = False) -> _AbstractLoad:

2505 if raiseload:

2506 kw = {"raiseload": raiseload}

2507 else:

2508 kw = {}

2509

2510 return _generate_from_keys(Load.defer, (key,), False, kw)

2511

2512

2513@loader_unbound_fn

2514def undefer(key: _AttrType) -> _AbstractLoad:

2515 return _generate_from_keys(Load.undefer, (key,), False, {})

2516

2517

2518@loader_unbound_fn

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

2520 element = _WildcardLoad()

2521 return element.undefer_group(name)

2522

2523

2524@loader_unbound_fn

2525def with_expression(

2526 key: _AttrType, expression: _ColumnExpressionArgument[Any]

2527) -> _AbstractLoad:

2528 return _generate_from_keys(

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

2530 )

2531

2532

2533@loader_unbound_fn

2534def selectin_polymorphic(

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

2536) -> _AbstractLoad:

2537 ul = Load(base_cls)

2538 return ul.selectin_polymorphic(classes)

2539

2540

2541def _raise_for_does_not_link(path, attrname, parent_entity):

2542 if len(path) > 1:

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

2544 if insp_is_aliased_class(parent_entity):

2545 parent_entity_str = str(parent_entity)

2546 else:

2547 parent_entity_str = parent_entity.class_.__name__

2548

2549 raise sa_exc.ArgumentError(

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

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

2552 % (

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

2554 (

2555 " Did you mean to use "

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

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

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

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

2560 if not path_is_of_type

2561 and not path[-1].is_aliased_class

2562 and orm_util._entity_corresponds_to(

2563 path.entity, inspect(parent_entity).mapper

2564 )

2565 else ""

2566 ),

2567 )

2568 )

2569 else:

2570 raise sa_exc.ArgumentError(

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

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

2573 )

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

788 statements