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 Literal

20from typing import Optional

21from typing import overload

22from typing import Sequence

23from typing import Tuple

24from typing import Type

25from typing import TypeVar

26from typing import Union

28from . import util as orm_util

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 entity_str

34from .base import InspectionAttr

35from .interfaces import LoaderOption

36from .path_registry import _AbstractEntityRegistry

37from .path_registry import _ACCEPTED_TOKENS

38from .path_registry import _COLUMN_TOKEN

39from .path_registry import _DEFAULT_TOKEN

40from .path_registry import _RELATIONSHIP_TOKEN

41from .path_registry import _StrPathToken

42from .path_registry import _TokenRegistry

43from .path_registry import _WILDCARD_TOKEN

44from .path_registry import path_is_property

45from .path_registry import PathRegistry

46from .util import _orm_full_deannotate

47from .util import AliasedInsp

48from .. import exc as sa_exc

49from .. import inspect

50from .. import util

51from ..sql import and_

52from ..sql import cache_key

53from ..sql import coercions

54from ..sql import roles

55from ..sql import traversals

56from ..sql import visitors

57from ..sql.base import _generative

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, ...]

86# maps _StrategyKey tuples to the user-facing loader function name,

87# populated by the @_strategy_labels() decorator below

88_STRATEGY_FN_LABELS: Dict[Any, str] = {}

91class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):

92 __slots__ = ("propagate_to_loaders",)

94 _is_strategy_option = True

95 propagate_to_loaders: bool

97 def contains_eager(

98 self,

99 attr: _AttrType,

100 alias: Optional[_FromClauseArgument] = None,

101 _is_chain: bool = False,

102 _propagate_to_loaders: bool = False,

103 ) -> Self:

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

105 columns stated manually in the query.

106

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

108 both method-chained and standalone operation.

109

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

111 the desired rows, i.e.::

112

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

114

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

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

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

118

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

120 collection; queries will normally want to use the

121 :ref:`orm_queryguide_populate_existing` execution option assuming the

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

123

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

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

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

127

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

129

130 .. seealso::

131

132 :ref:`loading_toplevel`

133

134 :ref:`contains_eager`

135

136 """

137 if alias is not None:

138 if not isinstance(alias, str):

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

140 else:

141 util.warn_deprecated(

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

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

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

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

146 version="1.4",

147 )

148 coerced_alias = alias

149

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

151 assert isinstance(attr, QueryableAttribute)

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

153 assert ot is not None

154 coerced_alias = ot.selectable

155 else:

156 coerced_alias = None

157

158 cloned = self._set_relationship_strategy(

159 attr,

160 {"lazy": "joined"},

161 propagate_to_loaders=_propagate_to_loaders,

162 opts={"eager_from_alias": coerced_alias},

163 _reconcile_to_other=True if _is_chain else None,

164 )

165 return cloned

166

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

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

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

170 deferred.

171

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

173 both method-chained and standalone operation.

174

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

176 ``fullname`` attributes::

177

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

179

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

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

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

183

184 session.query(User).options(

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

186 )

187

188 For a statement that has multiple entities,

189 the lead entity can be

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

191

192 stmt = (

193 select(User, Address)

194 .join(User.addresses)

195 .options(

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

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

198 )

199 )

200

201 When used together with the

202 :ref:`populate_existing <orm_queryguide_populate_existing>`

203 execution option only the attributes listed will be refreshed.

204

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

206

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

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

209 to prevent unwanted SQL from being emitted.

210

211 .. versionadded:: 2.0

212

213 .. seealso::

214

215 :ref:`orm_queryguide_column_deferral` - in the

216 :ref:`queryguide_toplevel`

217

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

219

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

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

222 to prevent unwanted SQL from being emitted.

223

224 .. versionadded:: 2.0

225

226 """

227 cloned = self._set_column_strategy(

228 _expand_column_strategy_attrs(attrs),

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

230 )

231

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

233 if raiseload:

234 wildcard_strategy["raiseload"] = True

235

236 cloned = cloned._set_column_strategy(

237 ("*",),

238 wildcard_strategy,

239 )

240 return cloned

241

242 def joinedload(

243 self,

244 attr: _AttrType,

245 innerjoin: Optional[bool] = None,

246 ) -> Self:

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

248 eager loading.

249

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

251 both method-chained and standalone operation.

252

253 examples::

254

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

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

257

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

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

260

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

262 # joined-load the keywords collection

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

264

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

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

267

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

269

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

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

272

273 select(A).options(

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

275 )

276

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

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

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

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

281 directly supported.

282

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

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

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

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

287 is an outerjoin::

288

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

290

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

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

293

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

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

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

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

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

299

300 .. note::

301

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

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

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

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

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

307

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

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

310 explicit JOINs with eager loading of collections, use

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

312

313 .. seealso::

314

315 :ref:`loading_toplevel`

316

317 :ref:`joined_eager_loading`

318

319 """ # noqa: E501

320 loader = self._set_relationship_strategy(

321 attr,

322 {"lazy": "joined"},

323 opts=(

324 {"innerjoin": innerjoin}

325 if innerjoin is not None

326 else util.EMPTY_DICT

327 ),

328 )

329 return loader

330

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

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

333 subquery eager loading.

334

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

336 both method-chained and standalone operation.

337

338 examples::

339

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

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

342

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

344 select(Order).options(

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

346 )

347

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

349 # subquery-load the keywords collection

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

351

352 .. seealso::

353

354 :ref:`loading_toplevel`

355

356 :ref:`subquery_eager_loading`

357

358 """

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

360

361 def selectinload(

362 self,

363 attr: _AttrType,

364 recursion_depth: Optional[int] = None,

365 chunksize: Optional[int] = None,

366 ) -> Self:

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

368 SELECT IN eager loading.

369

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

371 both method-chained and standalone operation.

372

373 examples::

374

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

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

377

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

379 select(Order).options(

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

381 )

382

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

384 # selectin-load the keywords collection

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

386

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

388 in conjunction with a self-referential relationship,

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

390 automatically until no items are found.

391

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

393 currently supports only self-referential relationships. There

394 is not yet an option to automatically traverse recursive structures

395 with more than one relationship involved.

396

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

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

399 status for the 2.0 series.

400

401 .. versionadded:: 2.0 added

402 :paramref:`_orm.selectinload.recursion_depth`

403

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

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

406 to the passed parameter

407

408 .. versionadded:: 2.1.0b3

409

410 .. seealso::

411

412 :ref:`loading_toplevel`

413

414 :ref:`selectin_eager_loading`

415

416 """

417 return self._set_relationship_strategy(

418 attr,

419 {"lazy": "selectin"},

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

421 )

422

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

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

425 loading.

426

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

428 both method-chained and standalone operation.

429

430 .. seealso::

431

432 :ref:`loading_toplevel`

433

434 :ref:`lazy_loading`

435

436 """

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

438

439 def immediateload(

440 self,

441 attr: _AttrType,

442 recursion_depth: Optional[int] = None,

443 ) -> Self:

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

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

446

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

448 fire off any additional eager loaders.

449

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

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

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

453

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

455 both method-chained and standalone operation.

456

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

458 in conjunction with a self-referential relationship,

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

460 automatically until no items are found.

461

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

463 currently supports only self-referential relationships. There

464 is not yet an option to automatically traverse recursive structures

465 with more than one relationship involved.

466

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

468 treated as "alpha" status

469

470 .. versionadded:: 2.0 added

471 :paramref:`_orm.immediateload.recursion_depth`

472

473

474 .. seealso::

475

476 :ref:`loading_toplevel`

477

478 :ref:`selectin_eager_loading`

479

480 """

481 loader = self._set_relationship_strategy(

482 attr,

483 {"lazy": "immediate"},

484 opts={"recursion_depth": recursion_depth},

485 )

486 return loader

487

488 @util.deprecated(

489 "2.1",

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

491 "in a future release. This option "

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

493 "items.",

494 )

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

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

497 unloaded.

498

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

500 producing any loading effect.

501

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

503 only.

504

505 .. seealso::

506

507 :ref:`loading_toplevel`

508

509 """

510

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

512

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

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

515

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

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

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

519 ensure that all relationship attributes that are accessed in a

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

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

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

523

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

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

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

527 :func:`.defer` loader option.

528

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

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

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

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

533

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

535 both method-chained and standalone operation.

536

537 .. seealso::

538

539 :ref:`loading_toplevel`

540

541 :ref:`prevent_lazy_with_raiseload`

542

543 :ref:`orm_queryguide_deferred_raiseload`

544

545 """

546

547 return self._set_relationship_strategy(

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

549 )

550

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

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

553

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

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

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

557 loading will be used.

558

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

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

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

562 element of an element::

563

564 session.query(MyClass).options(

565 defaultload(MyClass.someattribute).joinedload(

566 MyOtherClass.someotherattribute

567 )

568 )

569

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

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

572

573 session.scalars(

574 select(MyClass).options(

575 defaultload(MyClass.someattribute)

576 .defer("some_column")

577 .undefer("some_other_column")

578 )

579 )

580

581 .. seealso::

582

583 :ref:`orm_queryguide_relationship_sub_options`

584

585 :meth:`_orm.Load.options`

586

587 """

588 return self._set_relationship_strategy(attr, None)

589

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

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

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

593

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

595 both method-chained and standalone operation.

596

597 e.g.::

598

599 from sqlalchemy.orm import defer

600

601 session.query(MyClass).options(

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

603 )

604

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

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

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

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

609

610 session.query(MyClass).options(

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

612 )

613

614 Multiple deferral options related to a relationship can be bundled

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

616

617

618 select(MyClass).options(

619 defaultload(MyClass.someattr).options(

620 defer(RelatedClass.some_column),

621 defer(RelatedClass.some_other_column),

622 defer(RelatedClass.another_column),

623 )

624 )

625

626 :param key: Attribute to be deferred.

627

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

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

630 to prevent unwanted SQL from being emitted.

631

632 .. versionadded:: 1.4

633

634 .. seealso::

635

636 :ref:`orm_queryguide_column_deferral` - in the

637 :ref:`queryguide_toplevel`

638

639 :func:`_orm.load_only`

640

641 :func:`_orm.undefer`

642

643 """

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

645 if raiseload:

646 strategy["raiseload"] = True

647 return self._set_column_strategy(

648 _expand_column_strategy_attrs((key,)), strategy

649 )

650

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

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

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

654 as a whole.

655

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

657 :func:`.deferred` attribute.

658

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

660 both method-chained and standalone operation.

661

662 Examples::

663

664 # undefer two columns

665 session.query(MyClass).options(

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

667 )

668

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

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

671

672 # undefer a column on a related object

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

674

675 :param key: Attribute to be undeferred.

676

677 .. seealso::

678

679 :ref:`orm_queryguide_column_deferral` - in the

680 :ref:`queryguide_toplevel`

681

682 :func:`_orm.defer`

683

684 :func:`_orm.undefer_group`

685

686 """ # noqa: E501

687 return self._set_column_strategy(

688 _expand_column_strategy_attrs((key,)),

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

690 )

691

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

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

694 undeferred.

695

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

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

698

699 E.g::

700

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

702

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

704 spelled out using relationship loader options, such as

705 :func:`_orm.defaultload`::

706

707 select(MyClass).options(

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

709 )

710

711 .. seealso::

712

713 :ref:`orm_queryguide_column_deferral` - in the

714 :ref:`queryguide_toplevel`

715

716 :func:`_orm.defer`

717

718 :func:`_orm.undefer`

719

720 """

721 return self._set_column_strategy(

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

723 )

724

725 def with_expression(

726 self,

727 key: _AttrType,

728 expression: _ColumnExpressionArgument[Any],

729 ) -> Self:

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

731 attribute.

732

733 This option is used in conjunction with the

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

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

736

737 E.g.::

738

739 stmt = select(SomeClass).options(

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

741 )

742

743 :param key: Attribute to be populated

744

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

746

747 .. seealso::

748

749 :ref:`orm_queryguide_with_expression` - background and usage

750 examples

751

752 """

753

754 expression = _orm_full_deannotate(

755 coercions.expect(roles.LabeledColumnExprRole, expression)

756 )

757

758 return self._set_column_strategy(

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

760 )

761

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

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

764 specific to a subclass.

765

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

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

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

769

770 .. seealso::

771

772 :ref:`polymorphic_selectin`

773

774 """

775 self = self._set_class_strategy(

776 {"selectinload_polymorphic": True},

777 opts={

778 "entities": tuple(

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

780 )

781 },

782 )

783 return self

784

785 @overload

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

787

788 @overload

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

790

791 def _coerce_strat(

792 self, strategy: Optional[_StrategySpec]

793 ) -> Optional[_StrategyKey]:

794 if strategy is not None:

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

796 else:

797 strategy_key = None

798 return strategy_key

799

800 @_generative

801 def _set_relationship_strategy(

802 self,

803 attr: _AttrType,

804 strategy: Optional[_StrategySpec],

805 propagate_to_loaders: bool = True,

806 opts: Optional[_OptsType] = None,

807 _reconcile_to_other: Optional[bool] = None,

808 ) -> Self:

809 strategy_key = self._coerce_strat(strategy)

810

811 self._clone_for_bind_strategy(

812 (attr,),

813 strategy_key,

814 _RELATIONSHIP_TOKEN,

815 opts=opts,

816 propagate_to_loaders=propagate_to_loaders,

817 reconcile_to_other=_reconcile_to_other,

818 )

819 return self

820

821 @_generative

822 def _set_column_strategy(

823 self,

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

825 strategy: Optional[_StrategySpec],

826 opts: Optional[_OptsType] = None,

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

828 ) -> Self:

829 strategy_key = self._coerce_strat(strategy)

830

831 self._clone_for_bind_strategy(

832 attrs,

833 strategy_key,

834 _COLUMN_TOKEN,

835 opts=opts,

836 attr_group=attrs,

837 extra_criteria=extra_criteria,

838 )

839 return self

840

841 @_generative

842 def _set_generic_strategy(

843 self,

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

845 strategy: _StrategySpec,

846 _reconcile_to_other: Optional[bool] = None,

847 ) -> Self:

848 strategy_key = self._coerce_strat(strategy)

849 self._clone_for_bind_strategy(

850 attrs,

851 strategy_key,

852 None,

853 propagate_to_loaders=True,

854 reconcile_to_other=_reconcile_to_other,

855 )

856 return self

857

858 @_generative

859 def _set_class_strategy(

860 self, strategy: _StrategySpec, opts: _OptsType

861 ) -> Self:

862 strategy_key = self._coerce_strat(strategy)

863

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

865 return self

866

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

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

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

870

871 Implementation is provided by subclasses.

872

873 """

874 raise NotImplementedError()

875

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

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

878 :class:`_orm._AbstractLoad` object.

879

880 Implementation is provided by subclasses.

881

882 """

883 raise NotImplementedError()

884

885 def _clone_for_bind_strategy(

886 self,

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

888 strategy: Optional[_StrategyKey],

889 wildcard_key: Optional[_WildcardKeyType],

890 opts: Optional[_OptsType] = None,

891 attr_group: Optional[_AttrGroupType] = None,

892 propagate_to_loaders: bool = True,

893 reconcile_to_other: Optional[bool] = None,

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

895 ) -> Self:

896 raise NotImplementedError()

897

898 def process_compile_state_replaced_entities(

899 self,

900 compile_state: _ORMCompileState,

901 mapper_entities: Sequence[_MapperEntity],

902 ) -> None:

903 if not compile_state.compile_options._enable_eagerloads:

904 return

905

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

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

908 # for the entities having been replaced with equivalents

909 self._process(

910 compile_state,

911 mapper_entities,

912 not bool(compile_state.current_path),

913 )

914

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

916 if not compile_state.compile_options._enable_eagerloads:

917 return

918

919 self._process(

920 compile_state,

921 compile_state._lead_mapper_entities,

922 not bool(compile_state.current_path)

923 and not compile_state.compile_options._for_refresh_state,

924 )

925

926 def _process(

927 self,

928 compile_state: _ORMCompileState,

929 mapper_entities: Sequence[_MapperEntity],

930 raiseerr: bool,

931 ) -> None:

932 """implemented by subclasses"""

933 raise NotImplementedError()

934

935 @classmethod

936 def _chop_path(

937 cls,

938 to_chop: _PathRepresentation,

939 path: PathRegistry,

940 debug: bool = False,

941 ) -> Optional[_PathRepresentation]:

942 i = -1

943

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

945 zip(to_chop, path.natural_path)

946 ):

947 if isinstance(c_token, str):

948 if i == 0 and (

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

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

951 ):

952 return to_chop

953 elif (

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

955 and c_token != p_token.key # type: ignore[union-attr]

956 ):

957 return None

958

959 if c_token is p_token:

960 continue

961 elif (

962 isinstance(c_token, InspectionAttr)

963 and insp_is_mapper(c_token)

964 and insp_is_mapper(p_token)

965 and c_token.isa(p_token)

966 ):

967 continue

968

969 else:

970 return None

971 return to_chop[i + 1 :]

972

973

974class Load(_AbstractLoad):

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

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

977 order to affect how various mapped attributes are loaded.

978

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

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

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

982 except for in some very specific cases.

983

984 .. seealso::

985

986 :ref:`orm_queryguide_relationship_per_entity_wildcard` - illustrates an

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

988

989 """

990

991 __slots__ = (

992 "path",

993 "context",

994 "additional_source_entities",

995 )

996

997 _traverse_internals = [

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

999 (

1000 "context",

1001 visitors.InternalTraversal.dp_has_cache_key_list,

1002 ),

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

1004 (

1005 "additional_source_entities",

1006 visitors.InternalTraversal.dp_has_cache_key_list,

1007 ),

1008 ]

1009 _cache_key_traversal = None

1010

1011 path: PathRegistry

1012 context: Tuple[_LoadElement, ...]

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

1014

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

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

1017 insp._post_inspect

1018

1019 self.path = insp._path_registry

1020 self.context = ()

1021 self.propagate_to_loaders = False

1022 self.additional_source_entities = ()

1023

1024 def __str__(self) -> str:

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

1026

1027 @classmethod

1028 def _construct_for_existing_path(

1029 cls, path: _AbstractEntityRegistry

1030 ) -> Load:

1031 load = cls.__new__(cls)

1032 load.path = path

1033 load.context = ()

1034 load.propagate_to_loaders = False

1035 load.additional_source_entities = ()

1036 return load

1037

1038 def _adapt_cached_option_to_uncached_option(

1039 self, context: QueryContext, uncached_opt: ORMOption

1040 ) -> ORMOption:

1041 if uncached_opt is self:

1042 return self

1043 return self._adjust_for_extra_criteria(context)

1044

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

1046 cloned = self._clone()

1047 cloned.context = tuple(

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

1049 )

1050 return cloned

1051

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

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

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

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

1056

1057 """

1058

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

1060 # actually have any extra_criteria options, which is the

1061 # common case

1062 for value in self.context:

1063 if value._extra_criteria:

1064 break

1065 else:

1066 return self

1067

1068 replacement_cache_key = context.user_passed_query._generate_cache_key()

1069

1070 if replacement_cache_key is None:

1071 return self

1072

1073 orig_query = context.compile_state.select_statement

1074 orig_cache_key = orig_query._generate_cache_key()

1075 assert orig_cache_key is not None

1076

1077 def process(

1078 opt: _LoadElement,

1079 replacement_cache_key: CacheKey,

1080 orig_cache_key: CacheKey,

1081 ) -> _LoadElement:

1082 cloned_opt = opt._clone()

1083

1084 cloned_opt._extra_criteria = tuple(

1085 replacement_cache_key._apply_params_to_element(

1086 orig_cache_key, crit

1087 )

1088 for crit in cloned_opt._extra_criteria

1089 )

1090

1091 return cloned_opt

1092

1093 cloned = self._clone()

1094 cloned.context = tuple(

1095 (

1096 process(value, replacement_cache_key, orig_cache_key)

1097 if value._extra_criteria

1098 else value

1099 )

1100 for value in self.context

1101 )

1102 return cloned

1103

1104 def _reconcile_query_entities_with_us(self, mapper_entities, raiseerr):

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

1106 entity inside of _LoadElement objects.

1107

1108 """

1109 path = self.path

1110

1111 for ent in mapper_entities:

1112 ezero = ent.entity_zero

1113 if ezero and orm_util._entity_corresponds_to(

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

1115 # safe to pass to _entity_corresponds_to()

1116 ezero,

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

1118 ):

1119 return ezero

1120

1121 return None

1122

1123 def _process(

1124 self,

1125 compile_state: _ORMCompileState,

1126 mapper_entities: Sequence[_MapperEntity],

1127 raiseerr: bool,

1128 ) -> None:

1129 reconciled_lead_entity = self._reconcile_query_entities_with_us(

1130 mapper_entities, raiseerr

1131 )

1132

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

1134 has_current_path = bool(compile_state.compile_options._current_path)

1135

1136 for loader in self.context:

1137 # issue #11292

1138 # historically, propagate_to_loaders was only considered at

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

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

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

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

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

1144 # so we check again

1145 if has_current_path and not loader.propagate_to_loaders:

1146 continue

1147 loader.process_compile_state(

1148 self,

1149 compile_state,

1150 mapper_entities,

1151 reconciled_lead_entity,

1152 raiseerr,

1153 )

1154

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

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

1157 :class:`_orm.Load` object.

1158

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

1160

1161 """

1162 cloned = self._generate()

1163

1164 assert cloned.propagate_to_loaders == self.propagate_to_loaders

1165

1166 if not any(

1167 orm_util._entity_corresponds_to_use_path_impl(

1168 elem, cloned.path.odd_element(0)

1169 )

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

1171 + parent.additional_source_entities

1172 ):

1173 if len(cloned.path) > 1:

1174 attrname = cloned.path[1]

1175 parent_entity = cloned.path[0]

1176 else:

1177 attrname = cloned.path[0]

1178 parent_entity = cloned.path[0]

1179 _raise_for_does_not_link(parent.path, attrname, parent_entity)

1180

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

1182

1183 if self.context:

1184 cloned.context = tuple(

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

1186 )

1187

1188 if cloned.context:

1189 parent.context += cloned.context

1190 parent.additional_source_entities += (

1191 cloned.additional_source_entities

1192 )

1193

1194 @_generative

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

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

1197 :class:`_orm.Load`

1198 object.

1199

1200 E.g.::

1201

1202 query = session.query(Author)

1203 query = query.options(

1204 joinedload(Author.book).options(

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

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

1207 )

1208 )

1209

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

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

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

1213

1214 .. seealso::

1215

1216 :func:`.defaultload`

1217

1218 :ref:`orm_queryguide_relationship_sub_options`

1219

1220 """

1221 for opt in opts:

1222 try:

1223 opt._apply_to_parent(self)

1224 except AttributeError as ae:

1225 if not isinstance(opt, _AbstractLoad):

1226 raise sa_exc.ArgumentError(

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

1228 "Load.options() method."

1229 ) from ae

1230 else:

1231 raise

1232 return self

1233

1234 def _clone_for_bind_strategy(

1235 self,

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

1237 strategy: Optional[_StrategyKey],

1238 wildcard_key: Optional[_WildcardKeyType],

1239 opts: Optional[_OptsType] = None,

1240 attr_group: Optional[_AttrGroupType] = None,

1241 propagate_to_loaders: bool = True,

1242 reconcile_to_other: Optional[bool] = None,

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

1244 ) -> Self:

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

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

1247 # InstanceState.load_options

1248 if propagate_to_loaders:

1249 self.propagate_to_loaders = True

1250

1251 if self.path.is_token:

1252 raise sa_exc.ArgumentError(

1253 "Wildcard token cannot be followed by another entity"

1254 )

1255

1256 elif path_is_property(self.path):

1257 # reuse the lookup which will raise a nicely formatted

1258 # LoaderStrategyException

1259 if strategy:

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

1261 else:

1262 raise sa_exc.ArgumentError(

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

1264 "refer to a mapped entity"

1265 )

1266

1267 if attrs is None:

1268 load_element = _ClassStrategyLoad.create(

1269 self.path,

1270 None,

1271 strategy,

1272 wildcard_key,

1273 opts,

1274 propagate_to_loaders,

1275 attr_group=attr_group,

1276 reconcile_to_other=reconcile_to_other,

1277 extra_criteria=extra_criteria,

1278 )

1279 if load_element:

1280 self.context += (load_element,)

1281 assert opts is not None

1282 self.additional_source_entities += cast(

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

1284 )

1285

1286 else:

1287 for attr in attrs:

1288 if isinstance(attr, str):

1289 load_element = _TokenStrategyLoad.create(

1290 self.path,

1291 attr,

1292 strategy,

1293 wildcard_key,

1294 opts,

1295 propagate_to_loaders,

1296 attr_group=attr_group,

1297 reconcile_to_other=reconcile_to_other,

1298 extra_criteria=extra_criteria,

1299 )

1300 else:

1301 load_element = _AttributeStrategyLoad.create(

1302 self.path,

1303 attr,

1304 strategy,

1305 wildcard_key,

1306 opts,

1307 propagate_to_loaders,

1308 attr_group=attr_group,

1309 reconcile_to_other=reconcile_to_other,

1310 extra_criteria=extra_criteria,

1311 )

1312

1313 if load_element:

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

1315 # object with the latest path.

1316 if wildcard_key is _RELATIONSHIP_TOKEN:

1317 self.path = load_element.path

1318 self.context += (load_element,)

1319

1320 # this seems to be effective for selectinloader,

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

1322 # but does not work for immediateloader, which still

1323 # must add additional options at load time

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

1325 r1 = load_element._recurse()

1326 self.context += (r1,)

1327

1328 return self

1329

1330 def __getstate__(self):

1331 d = self._shallow_to_dict()

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

1333 return d

1334

1335 def __setstate__(self, state):

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

1337 self._shallow_from_dict(state)

1338

1339

1340class _WildcardLoad(_AbstractLoad):

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

1342

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

1344

1345 _traverse_internals = [

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

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

1348 (

1349 "local_opts",

1350 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1351 ),

1352 ]

1353 cache_key_traversal: _CacheKeyTraversalType = None

1354

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

1356 local_opts: _OptsType

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

1358 propagate_to_loaders = False

1359

1360 def __init__(self) -> None:

1361 self.path = ()

1362 self.strategy = None

1363 self.local_opts = util.EMPTY_DICT

1364

1365 def _clone_for_bind_strategy(

1366 self,

1367 attrs,

1368 strategy,

1369 wildcard_key,

1370 opts=None,

1371 attr_group=None,

1372 propagate_to_loaders=True,

1373 reconcile_to_other=None,

1374 extra_criteria=None,

1375 ):

1376 assert attrs is not None

1377 attr = attrs[0]

1378 assert (

1379 wildcard_key

1380 and isinstance(attr, str)

1381 and attr in (_WILDCARD_TOKEN, _DEFAULT_TOKEN)

1382 )

1383

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

1385

1386 self.strategy = strategy

1387 self.path = (attr,)

1388 if opts:

1389 self.local_opts = util.immutabledict(opts)

1390

1391 assert extra_criteria is None

1392

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

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

1395

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

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

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

1399

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

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

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

1403

1404 """

1405 assert self.path

1406 attr = self.path[0]

1407 if attr.endswith(_DEFAULT_TOKEN):

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

1409

1410 effective_path = cast(_AbstractEntityRegistry, parent.path).token(attr)

1411

1412 assert effective_path.is_token

1413

1414 loader = _TokenStrategyLoad.create(

1415 effective_path,

1416 None,

1417 self.strategy,

1418 None,

1419 self.local_opts,

1420 self.propagate_to_loaders,

1421 )

1422

1423 parent.context += (loader,)

1424

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

1426 is_refresh = compile_state.compile_options._for_refresh_state

1427

1428 if is_refresh and not self.propagate_to_loaders:

1429 return

1430

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

1432 current_path = compile_state.current_path

1433

1434 start_path: _PathRepresentation = self.path

1435

1436 if current_path:

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

1438 # None back here

1439 new_path = self._chop_path(start_path, current_path)

1440 if new_path is None:

1441 return

1442

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

1444 # just returns it

1445 assert new_path == start_path

1446

1447 # start_path is a single-token tuple

1448 assert start_path and len(start_path) == 1

1449

1450 token = start_path[0]

1451 assert isinstance(token, str)

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

1453

1454 if not entity:

1455 return

1456

1457 path_element = entity

1458

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

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

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

1462 # tokens and populate into the Load().

1463

1464 assert isinstance(token, str)

1465 loader = _TokenStrategyLoad.create(

1466 path_element._path_registry,

1467 token,

1468 self.strategy,

1469 None,

1470 self.local_opts,

1471 self.propagate_to_loaders,

1472 raiseerr=raiseerr,

1473 )

1474 if not loader:

1475 return

1476

1477 assert loader.path.is_token

1478

1479 # don't pass a reconciled lead entity here

1480 loader.process_compile_state(

1481 self, compile_state, mapper_entities, None, raiseerr

1482 )

1483

1484 return loader

1485

1486 def _find_entity_basestring(

1487 self,

1488 entities: Iterable[_InternalEntityType[Any]],

1489 token: str,

1490 raiseerr: bool,

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

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

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

1494 if raiseerr:

1495 raise sa_exc.ArgumentError(

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

1497 f"loader option to multiple entities "

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

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

1500 f"""{

1501 ", ".join(

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

1503 for ent in entities

1504 )

1505 }."""

1506 )

1507 elif token.endswith(_DEFAULT_TOKEN):

1508 raiseerr = False

1509

1510 for ent in entities:

1511 # return only the first _MapperEntity when searching

1512 # based on string prop name. Ideally object

1513 # attributes are used to specify more exactly.

1514 return ent

1515 else:

1516 if raiseerr:

1517 raise sa_exc.ArgumentError(

1518 "Query has only expression-based entities - "

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

1520 )

1521 else:

1522 return None

1523

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

1525 d = self._shallow_to_dict()

1526 return d

1527

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

1529 self._shallow_from_dict(state)

1530

1531

1532class _LoadElement(

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

1534):

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

1536 and pass options to it.

1537

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

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

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

1541

1542 .. versionadded:: 2.0

1543

1544 """

1545

1546 __slots__ = (

1547 "path",

1548 "strategy",

1549 "propagate_to_loaders",

1550 "local_opts",

1551 "_extra_criteria",

1552 "_reconcile_to_other",

1553 )

1554 __visit_name__ = "load_element"

1555

1556 _traverse_internals = [

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

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

1559 (

1560 "local_opts",

1561 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1562 ),

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

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

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

1566 ]

1567 _cache_key_traversal = None

1568

1569 _extra_criteria: Tuple[Any, ...]

1570

1571 _reconcile_to_other: Optional[bool]

1572 strategy: Optional[_StrategyKey]

1573 path: PathRegistry

1574 propagate_to_loaders: bool

1575

1576 local_opts: util.immutabledict[str, Any]

1577

1578 is_token_strategy: bool

1579 is_class_strategy: bool

1580

1581 def __hash__(self) -> int:

1582 return id(self)

1583

1584 def __eq__(self, other):

1585 return traversals.compare(self, other)

1586

1587 @property

1588 def is_opts_only(self) -> bool:

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

1590

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

1592 cls = self.__class__

1593 s = cls.__new__(cls)

1594

1595 self._shallow_copy_to(s)

1596 return s

1597

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

1599 new = self._clone()

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

1601 return new

1602

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

1604 d = self._shallow_to_dict()

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

1606 return d

1607

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

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

1610 self._shallow_from_dict(state)

1611

1612 def _raise_for_no_match(self, parent_loader, mapper_entities):

1613 path = parent_loader.path

1614

1615 found_entities = False

1616 for ent in mapper_entities:

1617 ezero = ent.entity_zero

1618 if ezero:

1619 found_entities = True

1620 break

1621

1622 if not found_entities:

1623 raise sa_exc.ArgumentError(

1624 "Query has only expression-based entities; "

1625 f"attribute loader option {self._to_option_method_string()} "

1626 "can't be applied here."

1627 )

1628 else:

1629 raise sa_exc.ArgumentError(

1630 f"Mapped class {entity_str(path[0])} referenced in "

1631 f"option {self._to_option_method_string()} does not apply "

1632 f"to any of the root entities in this query, e.g. "

1633 f"""{

1634 ", ".join(

1635 entity_str(x.entity_zero)

1636 for x in mapper_entities if x.entity_zero

1637 )}. Please """

1638 "specify the full path "

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

1640 "attribute. "

1641 )

1642

1643 def _adjust_effective_path_for_current_path(

1644 self, effective_path: PathRegistry, current_path: PathRegistry

1645 ) -> Optional[PathRegistry]:

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

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

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

1649 current_path.

1650

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

1652

1653 .. sourcecode:: text

1654

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

1656

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

1658

1659 The adjusted path would be:

1660

1661 .. sourcecode:: text

1662

1663 Item -> keywords -> Keyword

1664

1665

1666 """

1667 chopped_start_path = Load._chop_path(

1668 effective_path.natural_path, current_path

1669 )

1670 if not chopped_start_path:

1671 return None

1672

1673 tokens_removed_from_start_path = len(effective_path) - len(

1674 chopped_start_path

1675 )

1676

1677 loader_lead_path_element = self.path[tokens_removed_from_start_path]

1678

1679 effective_path = PathRegistry.coerce(

1680 (loader_lead_path_element,) + chopped_start_path[1:]

1681 )

1682

1683 return effective_path

1684

1685 def _init_path(

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

1687 ):

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

1689 a new path.

1690

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

1692 a :class:`._LoadElement` object.

1693

1694 """

1695 raise NotImplementedError()

1696

1697 def _prepare_for_compile_state(

1698 self,

1699 parent_loader,

1700 compile_state,

1701 mapper_entities,

1702 reconciled_lead_entity,

1703 raiseerr,

1704 ):

1705 """implemented by subclasses."""

1706 raise NotImplementedError()

1707

1708 def process_compile_state(

1709 self,

1710 parent_loader,

1711 compile_state,

1712 mapper_entities,

1713 reconciled_lead_entity,

1714 raiseerr,

1715 ):

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

1717 _LoadElement.

1718

1719 """

1720 keys = self._prepare_for_compile_state(

1721 parent_loader,

1722 compile_state,

1723 mapper_entities,

1724 reconciled_lead_entity,

1725 raiseerr,

1726 )

1727 for key in keys:

1728 if key in compile_state.attributes:

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

1730 self, compile_state.attributes[key]

1731 )

1732 else:

1733 compile_state.attributes[key] = self

1734

1735 @classmethod

1736 def create(

1737 cls,

1738 path: PathRegistry,

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

1740 strategy: Optional[_StrategyKey],

1741 wildcard_key: Optional[_WildcardKeyType],

1742 local_opts: Optional[_OptsType],

1743 propagate_to_loaders: bool,

1744 raiseerr: bool = True,

1745 attr_group: Optional[_AttrGroupType] = None,

1746 reconcile_to_other: Optional[bool] = None,

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

1748 ) -> _LoadElement:

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

1750

1751 opt = cls.__new__(cls)

1752 opt.path = path

1753 opt.strategy = strategy

1754 opt.propagate_to_loaders = propagate_to_loaders

1755 opt.local_opts = (

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

1757 )

1758 opt._extra_criteria = ()

1759

1760 if reconcile_to_other is not None:

1761 opt._reconcile_to_other = reconcile_to_other

1762 elif strategy is None and not local_opts:

1763 opt._reconcile_to_other = True

1764 else:

1765 opt._reconcile_to_other = None

1766

1767 path = opt._init_path(

1768 path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

1769 )

1770

1771 if not path:

1772 return None # type: ignore[return-value]

1773

1774 assert opt.is_token_strategy == path.is_token

1775

1776 opt.path = path

1777 return opt

1778

1779 def __init__(self) -> None:

1780 raise NotImplementedError()

1781

1782 def _recurse(self) -> _LoadElement:

1783 cloned = self._clone()

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

1785

1786 return cloned

1787

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

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

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

1791 path.

1792

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

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

1795

1796 """

1797

1798 if not any(

1799 orm_util._entity_corresponds_to_use_path_impl(

1800 elem,

1801 self.path.odd_element(0),

1802 )

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

1804 + parent.additional_source_entities

1805 ):

1806 raise sa_exc.ArgumentError(

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

1808 f'from element "{entity_str(parent.path[-1])}".'

1809 )

1810

1811 return self._prepend_path(parent.path)

1812

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

1814 cloned = self._clone()

1815

1816 assert cloned.strategy == self.strategy

1817 assert cloned.local_opts == self.local_opts

1818 assert cloned.is_class_strategy == self.is_class_strategy

1819

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

1821

1822 return cloned

1823

1824 @staticmethod

1825 def _reconcile(

1826 replacement: _LoadElement, existing: _LoadElement

1827 ) -> _LoadElement:

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

1829 the context.attributes under the same key.

1830

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

1832 existing one

1833

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

1835

1836 """

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

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

1839 # see test_poly_loading.py

1840

1841 if replacement._reconcile_to_other:

1842 return existing

1843 elif replacement._reconcile_to_other is False:

1844 return replacement

1845 elif existing._reconcile_to_other:

1846 return replacement

1847 elif existing._reconcile_to_other is False:

1848 return existing

1849

1850 if existing is replacement:

1851 return replacement

1852 elif (

1853 existing.strategy == replacement.strategy

1854 and existing.local_opts == replacement.local_opts

1855 ):

1856 return replacement

1857 elif replacement.is_opts_only:

1858 existing = existing._clone()

1859 existing.local_opts = existing.local_opts.union(

1860 replacement.local_opts

1861 )

1862 existing._extra_criteria += replacement._extra_criteria

1863 return existing

1864 elif existing.is_opts_only:

1865 replacement = replacement._clone()

1866 replacement.local_opts = replacement.local_opts.union(

1867 existing.local_opts

1868 )

1869 replacement._extra_criteria += existing._extra_criteria

1870 return replacement

1871 elif replacement.path.is_token:

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

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

1874 # will raise as below

1875 return replacement

1876

1877 raise sa_exc.InvalidRequestError(

1878 f"Loader strategy replacement "

1879 f"{replacement._to_option_method_string()} is in conflict "

1880 f"with existing strategy {existing._to_option_method_string()}"

1881 )

1882

1883 def _to_option_method_string(self) -> str:

1884 """Return a string representation of this :class:`._LoadElement`

1885 as it would be written as a loader option method call, e.g.

1886 ``"joinedload(User.orders)"``.

1887

1888 """

1889 assert (

1890 self.strategy is not None

1891 ), "to_option_method_string() requires a strategy to be set"

1892

1893 for opt_key in self.local_opts:

1894 fn = _STRATEGY_FN_LABELS.get((self.strategy, opt_key))

1895 if fn:

1896 break

1897 else:

1898 fn = _STRATEGY_FN_LABELS.get((self.strategy, None))

1899

1900 assert (

1901 fn is not None

1902 ), f"No _STRATEGY_FN_LABELS entry for strategy {self.strategy!r}"

1903

1904 return f"{fn}({self.path.path_string()})"

1905

1906

1907class _AttributeStrategyLoad(_LoadElement):

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

1909

1910 e.g.::

1911

1912 joinedload(User.addresses)

1913 defer(Order.name)

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

1915

1916 """

1917

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

1919

1920 __visit_name__ = "attribute_strategy_load_element"

1921

1922 _traverse_internals = _LoadElement._traverse_internals + [

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

1924 (

1925 "_path_with_polymorphic_path",

1926 visitors.ExtendedInternalTraversal.dp_has_cache_key,

1927 ),

1928 ]

1929

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

1931 _path_with_polymorphic_path: Optional[PathRegistry]

1932

1933 is_class_strategy = False

1934 is_token_strategy = False

1935

1936 def _init_path(

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

1938 ):

1939 assert attr is not None

1940 self._of_type = None

1941 self._path_with_polymorphic_path = None

1942 insp, _, prop = _parse_attr_argument(attr)

1943

1944 if insp.is_property:

1945 # direct property can be sent from internal strategy logic

1946 # that sets up specific loaders, such as

1947 # emit_lazyload->_lazyload_reverse

1948 # prop = found_property = attr

1949 prop = attr

1950 path = path[prop]

1951

1952 if path.has_entity:

1953 path = path.entity_path

1954 return path

1955

1956 elif not insp.is_attribute:

1957 # should not reach here;

1958 assert False

1959

1960 # here we assume we have user-passed InstrumentedAttribute

1961 if not orm_util._entity_corresponds_to_use_path_impl(

1962 path[-1], attr.parent

1963 ):

1964 if raiseerr:

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

1966 raise sa_exc.ArgumentError(

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

1968 "loader option to multiple entities in the "

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

1970 )

1971 else:

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

1973 else:

1974 return None

1975

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

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

1978 # test_relationship_criteria.py::RelationshipCriteriaTest::

1979 # test_selectinload_nested_criteria[True] if an existing

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

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

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

1983 # poorly defined.

1984 if extra_criteria:

1985 assert not attr._extra_criteria

1986 self._extra_criteria = extra_criteria

1987 else:

1988 self._extra_criteria = attr._extra_criteria

1989

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

1991 ac = attr._of_type

1992 ext_info = inspect(ac)

1993 self._of_type = ext_info

1994

1995 self._path_with_polymorphic_path = path.entity_path[prop]

1996

1997 path = path[prop][ext_info]

1998

1999 else:

2000 path = path[prop]

2001

2002 if path.has_entity:

2003 path = path.entity_path

2004

2005 return path

2006

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

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

2009

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

2011 ensures that the polymorphic path information is correctly updated

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

2013 """

2014 cloned = super()._prepend_path(path)

2015

2016 # Also prepend the parent path to _path_with_polymorphic_path if

2017 # present

2018 if self._path_with_polymorphic_path is not None:

2019 cloned._path_with_polymorphic_path = PathRegistry.coerce(

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

2021 )

2022

2023 return cloned

2024

2025 def _generate_extra_criteria(self, context):

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

2027 immediate "extra_criteria" stored with this Load object.

2028

2029 Load objects are typically pulled from the cached version of

2030 the statement from a QueryContext. The statement currently being

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

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

2033 they handle this criteria for a result set.

2034

2035 """

2036

2037 assert (

2038 self._extra_criteria

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

2040

2041 orig_query = context.compile_state.select_statement

2042 current_query = context.query

2043

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

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

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

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

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

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

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

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

2052

2053 # if orig_query is current_query:

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

2055 # return and_(*self._extra_criteria)

2056

2057 k1 = orig_query._generate_cache_key()

2058 k2 = current_query._generate_cache_key()

2059

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

2061

2062 def _set_of_type_info(self, context, current_path):

2063 assert self._path_with_polymorphic_path

2064

2065 pwpi = self._of_type

2066 assert pwpi

2067 if not pwpi.is_aliased_class:

2068 pwpi = inspect(

2069 orm_util.AliasedInsp._with_polymorphic_factory(

2070 pwpi.mapper.base_mapper,

2071 (pwpi.mapper,),

2072 aliased=True,

2073 _use_mapper_path=True,

2074 )

2075 )

2076 start_path = self._path_with_polymorphic_path

2077 if current_path:

2078 new_path = self._adjust_effective_path_for_current_path(

2079 start_path, current_path

2080 )

2081 if new_path is None:

2082 return

2083 start_path = new_path

2084

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

2086 if key in context:

2087 existing_aliased_insp = context[key]

2088 this_aliased_insp = pwpi

2089 if existing_aliased_insp.mapper is not this_aliased_insp.mapper:

2090 raise sa_exc.InvalidRequestError(

2091 "Can't combine of_type() targets "

2092 f"{existing_aliased_insp} and {this_aliased_insp} "

2093 f"along the relationship {start_path[-1]} in "

2094 "the same query. To load multiple subclasses along a "

2095 "single relationship, use a single with_polymorphic() "

2096 "object that includes all of the target classes."

2097 )

2098 new_aliased_insp = existing_aliased_insp._merge_with(

2099 this_aliased_insp

2100 )

2101 context[key] = new_aliased_insp

2102 else:

2103 context[key] = pwpi

2104

2105 def _prepare_for_compile_state(

2106 self,

2107 parent_loader,

2108 compile_state,

2109 mapper_entities,

2110 reconciled_lead_entity,

2111 raiseerr,

2112 ):

2113 # _AttributeStrategyLoad

2114

2115 current_path = compile_state.current_path

2116 is_refresh = compile_state.compile_options._for_refresh_state

2117 assert not self.path.is_token

2118

2119 if is_refresh and not self.propagate_to_loaders:

2120 return []

2121

2122 if self._of_type:

2123 # apply additional with_polymorphic alias that may have been

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

2125 self._set_of_type_info(compile_state.attributes, current_path)

2126

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

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

2129 return []

2130

2131 if raiseerr and not reconciled_lead_entity:

2132 self._raise_for_no_match(parent_loader, mapper_entities)

2133

2134 if self.path.has_entity:

2135 effective_path = self.path.parent

2136 else:

2137 effective_path = self.path

2138

2139 if current_path:

2140 assert effective_path is not None

2141 effective_path = self._adjust_effective_path_for_current_path(

2142 effective_path, current_path

2143 )

2144 if effective_path is None:

2145 return []

2146

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

2148

2149 def __getstate__(self):

2150 d = super().__getstate__()

2151

2152 # can't pickle this. See

2153 # test_pickled.py -> test_lazyload_extra_criteria_not_supported

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

2155 # would be non-None

2156 d["_extra_criteria"] = ()

2157

2158 if self._path_with_polymorphic_path:

2159 d["_path_with_polymorphic_path"] = (

2160 self._path_with_polymorphic_path.serialize()

2161 )

2162

2163 if self._of_type:

2164 if self._of_type.is_aliased_class:

2165 d["_of_type"] = None

2166 elif self._of_type.is_mapper:

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

2168 else:

2169 assert False, "unexpected object for _of_type"

2170

2171 return d

2172

2173 def __setstate__(self, state):

2174 super().__setstate__(state)

2175

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

2177 self._path_with_polymorphic_path = PathRegistry.deserialize(

2178 state["_path_with_polymorphic_path"]

2179 )

2180 else:

2181 self._path_with_polymorphic_path = None

2182

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

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

2185 else:

2186 self._of_type = None

2187

2188

2189class _TokenStrategyLoad(_LoadElement):

2190 """Loader strategies against wildcard attributes

2191

2192 e.g.::

2193

2194 raiseload("*")

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

2196 defer("*")

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

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

2199

2200 """

2201

2202 __visit_name__ = "token_strategy_load_element"

2203

2204 inherit_cache = True

2205 is_class_strategy = False

2206 is_token_strategy = True

2207

2208 def _init_path(

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

2210 ):

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

2212 if attr is not None:

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

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

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

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

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

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

2219 # through to build a loader path that matches nothing

2220 if attr in _ACCEPTED_TOKENS:

2221 if wildcard_key:

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

2223

2224 path = path.token(attr)

2225 return path

2226 else:

2227 raise sa_exc.ArgumentError(

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

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

2230 )

2231 return path

2232

2233 def _prepare_for_compile_state(

2234 self,

2235 parent_loader,

2236 compile_state,

2237 mapper_entities,

2238 reconciled_lead_entity,

2239 raiseerr,

2240 ):

2241 # _TokenStrategyLoad

2242

2243 current_path = compile_state.current_path

2244 is_refresh = compile_state.compile_options._for_refresh_state

2245

2246 assert self.path.is_token

2247

2248 if is_refresh and not self.propagate_to_loaders:

2249 return []

2250

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

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

2253 return []

2254

2255 effective_path = self.path

2256 if reconciled_lead_entity:

2257 effective_path = PathRegistry.coerce(

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

2259 )

2260

2261 if current_path:

2262 new_effective_path = self._adjust_effective_path_for_current_path(

2263 effective_path, current_path

2264 )

2265 if new_effective_path is None:

2266 return []

2267 effective_path = new_effective_path

2268

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

2270 # to encompass everything from the query entity on

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

2272 # is set.

2273

2274 return [

2275 ("loader", natural_path)

2276 for natural_path in (

2277 cast(

2278 _TokenRegistry, effective_path

2279 )._generate_natural_for_superclasses()

2280 )

2281 ]

2282

2283

2284class _ClassStrategyLoad(_LoadElement):

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

2286 an attribute path

2287

2288 e.g.::

2289

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

2291 selectin_polymorphic(Person, [Engineer, Manager])

2292 )

2293

2294 """

2295

2296 inherit_cache = True

2297 is_class_strategy = True

2298 is_token_strategy = False

2299

2300 __visit_name__ = "class_strategy_load_element"

2301

2302 def _init_path(

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

2304 ):

2305 return path

2306

2307 def _prepare_for_compile_state(

2308 self,

2309 parent_loader,

2310 compile_state,

2311 mapper_entities,

2312 reconciled_lead_entity,

2313 raiseerr,

2314 ):

2315 # _ClassStrategyLoad

2316

2317 current_path = compile_state.current_path

2318 is_refresh = compile_state.compile_options._for_refresh_state

2319

2320 if is_refresh and not self.propagate_to_loaders:

2321 return []

2322

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

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

2325 return []

2326

2327 effective_path = self.path

2328

2329 if current_path:

2330 new_effective_path = self._adjust_effective_path_for_current_path(

2331 effective_path, current_path

2332 )

2333 if new_effective_path is None:

2334 return []

2335 effective_path = new_effective_path

2336

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

2338

2339

2340def _generate_from_keys(

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

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

2343 chained: bool,

2344 kw: Any,

2345) -> _AbstractLoad:

2346 lead_element: Optional[_AbstractLoad] = None

2347

2348 attr: Any

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

2350 for attr in _keys:

2351 if isinstance(attr, str):

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

2353 util.warn_deprecated(

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

2355 "deprecated "

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

2357 "it is "

2358 "believed to be unused. "

2359 "If you have been using this functionality, "

2360 "please "

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

2362 "tracker.",

2363 version="1.4",

2364 )

2365 attr = attr[1:]

2366

2367 if attr == _WILDCARD_TOKEN:

2368 if is_default:

2369 raise sa_exc.ArgumentError(

2370 "Wildcard token cannot be followed by "

2371 "another entity",

2372 )

2373

2374 if lead_element is None:

2375 lead_element = _WildcardLoad()

2376

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

2378

2379 else:

2380 raise sa_exc.ArgumentError(

2381 "Strings are not accepted for attribute names in "

2382 "loader options; please use class-bound "

2383 "attributes directly.",

2384 )

2385 else:

2386 if lead_element is None:

2387 _, lead_entity, _ = _parse_attr_argument(attr)

2388 lead_element = Load(lead_entity)

2389

2390 if is_default:

2391 if not chained:

2392 lead_element = lead_element.defaultload(attr)

2393 else:

2394 lead_element = meth(

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

2396 )

2397 else:

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

2399

2400 assert lead_element

2401 return lead_element

2402

2403

2404def _parse_attr_argument(

2405 attr: _AttrType,

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

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

2408 :class:`._AbstractLoad` instance.

2409

2410 This is used by the standalone loader strategy functions like

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

2412 :class:`._WildcardLoad` objects.

2413

2414 """

2415 try:

2416 insp: InspectionAttr = inspect(attr)

2417 except sa_exc.NoInspectionAvailable as err:

2418 raise sa_exc.ArgumentError(

2419 "expected ORM mapped attribute for loader strategy argument"

2420 ) from err

2421

2422 lead_entity: _InternalEntityType[Any]

2423

2424 if insp_is_mapper_property(insp):

2425 lead_entity = insp.parent

2426 prop = insp

2427 elif insp_is_attribute(insp):

2428 lead_entity = insp.parent

2429 prop = insp.prop

2430 else:

2431 raise sa_exc.ArgumentError(

2432 "expected ORM mapped attribute for loader strategy argument"

2433 )

2434

2435 return insp, lead_entity, prop

2436

2437

2438def _strategy_labels(

2439 *strategy_keys: "Any",

2440 discriminating_opt: Optional[str] = None,

2441) -> Callable[[_FN], _FN]:

2442 """Decorator that registers strategy key(s) -> function name in

2443 ``_STRATEGY_FN_LABELS``. Apply below ``@loader_unbound_fn`` so that

2444 ``fn.__name__`` is still the original function name when the decorator

2445 runs.

2446

2447 :param discriminating_opt: optional local_opts key that distinguishes

2448 this function from another function with the same strategy key.

2449 When set, the entry is stored under ``(strategy_key, opt_key)``

2450 instead of ``(strategy_key, None)``, and ``__str__`` will use this

2451 function name when that opt is present in ``local_opts``.

2452 """

2453

2454 def decorator(fn: _FN) -> _FN:

2455 for key in strategy_keys:

2456 _STRATEGY_FN_LABELS[(key, discriminating_opt)] = fn.__name__

2457 return fn

2458

2459 return decorator

2460

2461

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

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

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

2465

2466 """

2467 bound_fn = getattr(_AbstractLoad, fn.__name__)

2468 fn_doc = bound_fn.__doc__

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

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

2471

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

2473

2474"""

2475

2476 fn.__doc__ = fn_doc

2477 return fn

2478

2479

2480def _expand_column_strategy_attrs(

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

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

2483 return cast(

2484 "Tuple[_AttrType, ...]",

2485 tuple(

2486 a

2487 for attr in attrs

2488 for a in (

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

2490 if hasattr(attr, "_column_strategy_attrs")

2491 else (attr,)

2492 )

2493 ),

2494 )

2495

2496

2497# standalone functions follow. docstrings are filled in

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

2499

2500

2501@loader_unbound_fn

2502@_strategy_labels((("lazy", "joined"),), discriminating_opt="eager_from_alias")

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

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

2505

2506

2507@loader_unbound_fn

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

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

2510 # add some extra state to Load of some kind

2511 attrs = _expand_column_strategy_attrs(attrs)

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

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

2514

2515

2516@loader_unbound_fn

2517@_strategy_labels((("lazy", "joined"),))

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

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

2520

2521

2522@loader_unbound_fn

2523@_strategy_labels((("lazy", "subquery"),))

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

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

2526

2527

2528@loader_unbound_fn

2529@_strategy_labels((("lazy", "selectin"),))

2530def selectinload(

2531 *keys: _AttrType,

2532 recursion_depth: Optional[int] = None,

2533 chunksize: Optional[int] = None,

2534) -> _AbstractLoad:

2535 return _generate_from_keys(

2536 Load.selectinload,

2537 keys,

2538 False,

2539 {"recursion_depth": recursion_depth, "chunksize": chunksize},

2540 )

2541

2542

2543@loader_unbound_fn

2544@_strategy_labels((("lazy", "select"),))

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

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

2547

2548

2549@loader_unbound_fn

2550@_strategy_labels((("lazy", "immediate"),))

2551def immediateload(

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

2553) -> _AbstractLoad:

2554 return _generate_from_keys(

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

2556 )

2557

2558

2559@loader_unbound_fn

2560@_strategy_labels((("lazy", "noload"),))

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

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

2563

2564

2565@loader_unbound_fn

2566@_strategy_labels((("lazy", "raise"),), (("lazy", "raise_on_sql"),))

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

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

2569

2570

2571@loader_unbound_fn

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

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

2574

2575

2576@loader_unbound_fn

2577@_strategy_labels(

2578 (("deferred", True), ("instrument", True)),

2579 (("deferred", True), ("instrument", True), ("raiseload", True)),

2580)

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

2582 if raiseload:

2583 kw = {"raiseload": raiseload}

2584 else:

2585 kw = {}

2586

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

2588

2589

2590@loader_unbound_fn

2591@_strategy_labels((("deferred", False), ("instrument", True)))

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

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

2594

2595

2596@loader_unbound_fn

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

2598 element = _WildcardLoad()

2599 return element.undefer_group(name)

2600

2601

2602@loader_unbound_fn

2603@_strategy_labels((("query_expression", True),))

2604def with_expression(

2605 key: _AttrType, expression: _ColumnExpressionArgument[Any]

2606) -> _AbstractLoad:

2607 return _generate_from_keys(

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

2609 )

2610

2611

2612@loader_unbound_fn

2613@_strategy_labels((("selectinload_polymorphic", True),))

2614def selectin_polymorphic(

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

2616) -> _AbstractLoad:

2617 ul = Load(base_cls)

2618 return ul.selectin_polymorphic(classes)

2619

2620

2621def _raise_for_does_not_link(path, attrname, parent_entity):

2622 if len(path) > 1:

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

2624

2625 raise sa_exc.ArgumentError(

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

2627 f'link from relationship "{entity_str(path[-2])}%s".%s'

2628 % (

2629 (

2630 f".of_type({entity_str(path[-1])})"

2631 if path_is_of_type

2632 else ""

2633 ),

2634 (

2635 " Did you mean to use "

2636 f'"{entity_str(path[-2])}'

2637 f'.of_type({entity_str(parent_entity)})" or '

2638 '"loadopt.options('

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

2640 f'[{entity_str(parent_entity)}]), ...)" ?'

2641 if not path_is_of_type

2642 and not path[-1].is_aliased_class

2643 and orm_util._entity_corresponds_to(

2644 path.entity, inspect(parent_entity).mapper

2645 )

2646 else ""

2647 ),

2648 )

2649 )

2650 else:

2651 raise sa_exc.ArgumentError(

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

2653 f'link mapped class "{entity_str(path[-1])}"'

2654 )

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

814 statements