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_attribute

31from ._typing import insp_is_mapper

32from ._typing import insp_is_mapper_property

33from .attributes import QueryableAttribute

34from .base import entity_str

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

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

88# populated by the @_strategy_labels() decorator below

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

92class _AbstractLoad(traversals.GenerativeOnTraversal, LoaderOption):

93 __slots__ = ("propagate_to_loaders",)

95 _is_strategy_option = True

96 propagate_to_loaders: bool

98 def contains_eager(

99 self,

100 attr: _AttrType,

101 alias: Optional[_FromClauseArgument] = None,

102 _is_chain: bool = False,

103 _propagate_to_loaders: bool = False,

104 ) -> Self:

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

106 columns stated manually in the query.

107

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

109 both method-chained and standalone operation.

110

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

112 the desired rows, i.e.::

113

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

115

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

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

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

119

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

121 collection; queries will normally want to use the

122 :ref:`orm_queryguide_populate_existing` execution option assuming the

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

124

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

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

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

128

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

130

131 .. seealso::

132

133 :ref:`loading_toplevel`

134

135 :ref:`contains_eager`

136

137 """

138 if alias is not None:

139 if not isinstance(alias, str):

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

141 else:

142 util.warn_deprecated(

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

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

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

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

147 version="1.4",

148 )

149 coerced_alias = alias

150

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

152 assert isinstance(attr, QueryableAttribute)

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

154 assert ot is not None

155 coerced_alias = ot.selectable

156 else:

157 coerced_alias = None

158

159 cloned = self._set_relationship_strategy(

160 attr,

161 {"lazy": "joined"},

162 propagate_to_loaders=_propagate_to_loaders,

163 opts={"eager_from_alias": coerced_alias},

164 _reconcile_to_other=True if _is_chain else None,

165 )

166 return cloned

167

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

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

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

171 deferred.

172

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

174 both method-chained and standalone operation.

175

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

177 ``fullname`` attributes::

178

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

180

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

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

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

184

185 session.query(User).options(

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

187 )

188

189 For a statement that has multiple entities,

190 the lead entity can be

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

192

193 stmt = (

194 select(User, Address)

195 .join(User.addresses)

196 .options(

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

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

199 )

200 )

201

202 When used together with the

203 :ref:`populate_existing <orm_queryguide_populate_existing>`

204 execution option only the attributes listed will be refreshed.

205

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

207

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

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

210 to prevent unwanted SQL from being emitted.

211

212 .. versionadded:: 2.0

213

214 .. seealso::

215

216 :ref:`orm_queryguide_column_deferral` - in the

217 :ref:`queryguide_toplevel`

218

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

220

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

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

223 to prevent unwanted SQL from being emitted.

224

225 .. versionadded:: 2.0

226

227 """

228 cloned = self._set_column_strategy(

229 _expand_column_strategy_attrs(attrs),

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

231 )

232

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

234 if raiseload:

235 wildcard_strategy["raiseload"] = True

236

237 cloned = cloned._set_column_strategy(

238 ("*",),

239 wildcard_strategy,

240 )

241 return cloned

242

243 def joinedload(

244 self,

245 attr: _AttrType,

246 innerjoin: Optional[bool] = None,

247 ) -> Self:

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

249 eager loading.

250

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

252 both method-chained and standalone operation.

253

254 examples::

255

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

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

258

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

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

261

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

263 # joined-load the keywords collection

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

265

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

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

268

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

270

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

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

273

274 select(A).options(

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

276 )

277

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

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

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

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

282 directly supported.

283

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

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

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

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

288 is an outerjoin::

289

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

291

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

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

294

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

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

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

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

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

300

301 .. note::

302

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

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

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

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

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

308

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

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

311 explicit JOINs with eager loading of collections, use

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

313

314 .. seealso::

315

316 :ref:`loading_toplevel`

317

318 :ref:`joined_eager_loading`

319

320 """ # noqa: E501

321 loader = self._set_relationship_strategy(

322 attr,

323 {"lazy": "joined"},

324 opts=(

325 {"innerjoin": innerjoin}

326 if innerjoin is not None

327 else util.EMPTY_DICT

328 ),

329 )

330 return loader

331

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

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

334 subquery eager loading.

335

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

337 both method-chained and standalone operation.

338

339 examples::

340

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

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

343

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

345 select(Order).options(

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

347 )

348

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

350 # subquery-load the keywords collection

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

352

353 .. seealso::

354

355 :ref:`loading_toplevel`

356

357 :ref:`subquery_eager_loading`

358

359 """

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

361

362 def selectinload(

363 self,

364 attr: _AttrType,

365 recursion_depth: Optional[int] = None,

366 chunksize: Optional[int] = None,

367 ) -> Self:

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

369 SELECT IN eager loading.

370

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

372 both method-chained and standalone operation.

373

374 examples::

375

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

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

378

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

380 select(Order).options(

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

382 )

383

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

385 # selectin-load the keywords collection

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

387

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

389 in conjunction with a self-referential relationship,

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

391 automatically until no items are found.

392

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

394 currently supports only self-referential relationships. There

395 is not yet an option to automatically traverse recursive structures

396 with more than one relationship involved.

397

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

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

400 status for the 2.0 series.

401

402 .. versionadded:: 2.0 added

403 :paramref:`_orm.selectinload.recursion_depth`

404

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

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

407 to the passed parameter

408

409 .. versionadded:: 2.1.0b3

410

411 .. seealso::

412

413 :ref:`loading_toplevel`

414

415 :ref:`selectin_eager_loading`

416

417 """

418 return self._set_relationship_strategy(

419 attr,

420 {"lazy": "selectin"},

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

422 )

423

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

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

426 loading.

427

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

429 both method-chained and standalone operation.

430

431 .. seealso::

432

433 :ref:`loading_toplevel`

434

435 :ref:`lazy_loading`

436

437 """

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

439

440 def immediateload(

441 self,

442 attr: _AttrType,

443 recursion_depth: Optional[int] = None,

444 ) -> Self:

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

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

447

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

449 fire off any additional eager loaders.

450

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

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

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

454

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

456 both method-chained and standalone operation.

457

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

459 in conjunction with a self-referential relationship,

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

461 automatically until no items are found.

462

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

464 currently supports only self-referential relationships. There

465 is not yet an option to automatically traverse recursive structures

466 with more than one relationship involved.

467

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

469 treated as "alpha" status

470

471 .. versionadded:: 2.0 added

472 :paramref:`_orm.immediateload.recursion_depth`

473

474

475 .. seealso::

476

477 :ref:`loading_toplevel`

478

479 :ref:`selectin_eager_loading`

480

481 """

482 loader = self._set_relationship_strategy(

483 attr,

484 {"lazy": "immediate"},

485 opts={"recursion_depth": recursion_depth},

486 )

487 return loader

488

489 @util.deprecated(

490 "2.1",

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

492 "in a future release. This option "

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

494 "items.",

495 )

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

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

498 unloaded.

499

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

501 producing any loading effect.

502

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

504 only.

505

506 .. seealso::

507

508 :ref:`loading_toplevel`

509

510 """

511

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

513

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

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

516

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

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

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

520 ensure that all relationship attributes that are accessed in a

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

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

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

524

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

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

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

528 :func:`.defer` loader option.

529

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

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

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

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

534

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

536 both method-chained and standalone operation.

537

538 .. seealso::

539

540 :ref:`loading_toplevel`

541

542 :ref:`prevent_lazy_with_raiseload`

543

544 :ref:`orm_queryguide_deferred_raiseload`

545

546 """

547

548 return self._set_relationship_strategy(

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

550 )

551

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

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

554

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

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

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

558 loading will be used.

559

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

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

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

563 element of an element::

564

565 session.query(MyClass).options(

566 defaultload(MyClass.someattribute).joinedload(

567 MyOtherClass.someotherattribute

568 )

569 )

570

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

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

573

574 session.scalars(

575 select(MyClass).options(

576 defaultload(MyClass.someattribute)

577 .defer("some_column")

578 .undefer("some_other_column")

579 )

580 )

581

582 .. seealso::

583

584 :ref:`orm_queryguide_relationship_sub_options`

585

586 :meth:`_orm.Load.options`

587

588 """

589 return self._set_relationship_strategy(attr, None)

590

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

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

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

594

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

596 both method-chained and standalone operation.

597

598 e.g.::

599

600 from sqlalchemy.orm import defer

601

602 session.query(MyClass).options(

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

604 )

605

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

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

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

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

610

611 session.query(MyClass).options(

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

613 )

614

615 Multiple deferral options related to a relationship can be bundled

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

617

618

619 select(MyClass).options(

620 defaultload(MyClass.someattr).options(

621 defer(RelatedClass.some_column),

622 defer(RelatedClass.some_other_column),

623 defer(RelatedClass.another_column),

624 )

625 )

626

627 :param key: Attribute to be deferred.

628

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

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

631 to prevent unwanted SQL from being emitted.

632

633 .. versionadded:: 1.4

634

635 .. seealso::

636

637 :ref:`orm_queryguide_column_deferral` - in the

638 :ref:`queryguide_toplevel`

639

640 :func:`_orm.load_only`

641

642 :func:`_orm.undefer`

643

644 """

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

646 if raiseload:

647 strategy["raiseload"] = True

648 return self._set_column_strategy(

649 _expand_column_strategy_attrs((key,)), strategy

650 )

651

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

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

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

655 as a whole.

656

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

658 :func:`.deferred` attribute.

659

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

661 both method-chained and standalone operation.

662

663 Examples::

664

665 # undefer two columns

666 session.query(MyClass).options(

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

668 )

669

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

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

672

673 # undefer a column on a related object

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

675

676 :param key: Attribute to be undeferred.

677

678 .. seealso::

679

680 :ref:`orm_queryguide_column_deferral` - in the

681 :ref:`queryguide_toplevel`

682

683 :func:`_orm.defer`

684

685 :func:`_orm.undefer_group`

686

687 """ # noqa: E501

688 return self._set_column_strategy(

689 _expand_column_strategy_attrs((key,)),

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

691 )

692

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

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

695 undeferred.

696

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

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

699

700 E.g::

701

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

703

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

705 spelled out using relationship loader options, such as

706 :func:`_orm.defaultload`::

707

708 select(MyClass).options(

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

710 )

711

712 .. seealso::

713

714 :ref:`orm_queryguide_column_deferral` - in the

715 :ref:`queryguide_toplevel`

716

717 :func:`_orm.defer`

718

719 :func:`_orm.undefer`

720

721 """

722 return self._set_column_strategy(

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

724 )

725

726 def with_expression(

727 self,

728 key: _AttrType,

729 expression: _ColumnExpressionArgument[Any],

730 ) -> Self:

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

732 attribute.

733

734 This option is used in conjunction with the

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

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

737

738 E.g.::

739

740 stmt = select(SomeClass).options(

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

742 )

743

744 :param key: Attribute to be populated

745

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

747

748 .. seealso::

749

750 :ref:`orm_queryguide_with_expression` - background and usage

751 examples

752

753 """

754

755 expression = _orm_full_deannotate(

756 coercions.expect(roles.LabeledColumnExprRole, expression)

757 )

758

759 return self._set_column_strategy(

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

761 )

762

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

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

765 specific to a subclass.

766

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

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

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

770

771 .. seealso::

772

773 :ref:`polymorphic_selectin`

774

775 """

776 self = self._set_class_strategy(

777 {"selectinload_polymorphic": True},

778 opts={

779 "entities": tuple(

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

781 )

782 },

783 )

784 return self

785

786 @overload

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

788

789 @overload

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

791

792 def _coerce_strat(

793 self, strategy: Optional[_StrategySpec]

794 ) -> Optional[_StrategyKey]:

795 if strategy is not None:

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

797 else:

798 strategy_key = None

799 return strategy_key

800

801 @_generative

802 def _set_relationship_strategy(

803 self,

804 attr: _AttrType,

805 strategy: Optional[_StrategySpec],

806 propagate_to_loaders: bool = True,

807 opts: Optional[_OptsType] = None,

808 _reconcile_to_other: Optional[bool] = None,

809 ) -> Self:

810 strategy_key = self._coerce_strat(strategy)

811

812 self._clone_for_bind_strategy(

813 (attr,),

814 strategy_key,

815 _RELATIONSHIP_TOKEN,

816 opts=opts,

817 propagate_to_loaders=propagate_to_loaders,

818 reconcile_to_other=_reconcile_to_other,

819 )

820 return self

821

822 @_generative

823 def _set_column_strategy(

824 self,

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

826 strategy: Optional[_StrategySpec],

827 opts: Optional[_OptsType] = None,

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

829 ) -> Self:

830 strategy_key = self._coerce_strat(strategy)

831

832 self._clone_for_bind_strategy(

833 attrs,

834 strategy_key,

835 _COLUMN_TOKEN,

836 opts=opts,

837 attr_group=attrs,

838 extra_criteria=extra_criteria,

839 )

840 return self

841

842 @_generative

843 def _set_generic_strategy(

844 self,

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

846 strategy: _StrategySpec,

847 _reconcile_to_other: Optional[bool] = None,

848 ) -> Self:

849 strategy_key = self._coerce_strat(strategy)

850 self._clone_for_bind_strategy(

851 attrs,

852 strategy_key,

853 None,

854 propagate_to_loaders=True,

855 reconcile_to_other=_reconcile_to_other,

856 )

857 return self

858

859 @_generative

860 def _set_class_strategy(

861 self, strategy: _StrategySpec, opts: _OptsType

862 ) -> Self:

863 strategy_key = self._coerce_strat(strategy)

864

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

866 return self

867

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

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

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

871

872 Implementation is provided by subclasses.

873

874 """

875 raise NotImplementedError()

876

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

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

879 :class:`_orm._AbstractLoad` object.

880

881 Implementation is provided by subclasses.

882

883 """

884 raise NotImplementedError()

885

886 def _clone_for_bind_strategy(

887 self,

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

889 strategy: Optional[_StrategyKey],

890 wildcard_key: Optional[_WildcardKeyType],

891 opts: Optional[_OptsType] = None,

892 attr_group: Optional[_AttrGroupType] = None,

893 propagate_to_loaders: bool = True,

894 reconcile_to_other: Optional[bool] = None,

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

896 ) -> Self:

897 raise NotImplementedError()

898

899 def process_compile_state_replaced_entities(

900 self,

901 compile_state: _ORMCompileState,

902 mapper_entities: Sequence[_MapperEntity],

903 ) -> None:

904 if not compile_state.compile_options._enable_eagerloads:

905 return

906

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

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

909 # for the entities having been replaced with equivalents

910 self._process(

911 compile_state,

912 mapper_entities,

913 not bool(compile_state.current_path),

914 )

915

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

917 if not compile_state.compile_options._enable_eagerloads:

918 return

919

920 self._process(

921 compile_state,

922 compile_state._lead_mapper_entities,

923 not bool(compile_state.current_path)

924 and not compile_state.compile_options._for_refresh_state,

925 )

926

927 def _process(

928 self,

929 compile_state: _ORMCompileState,

930 mapper_entities: Sequence[_MapperEntity],

931 raiseerr: bool,

932 ) -> None:

933 """implemented by subclasses"""

934 raise NotImplementedError()

935

936 @classmethod

937 def _chop_path(

938 cls,

939 to_chop: _PathRepresentation,

940 path: PathRegistry,

941 debug: bool = False,

942 ) -> Optional[_PathRepresentation]:

943 i = -1

944

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

946 zip(to_chop, path.natural_path)

947 ):

948 if isinstance(c_token, str):

949 if i == 0 and (

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

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

952 ):

953 return to_chop

954 elif (

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

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

957 ):

958 return None

959

960 if c_token is p_token:

961 continue

962 elif (

963 isinstance(c_token, InspectionAttr)

964 and insp_is_mapper(c_token)

965 and insp_is_mapper(p_token)

966 and c_token.isa(p_token)

967 ):

968 continue

969

970 else:

971 return None

972 return to_chop[i + 1 :]

973

974

975class Load(_AbstractLoad):

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

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

978 order to affect how various mapped attributes are loaded.

979

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

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

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

983 except for in some very specific cases.

984

985 .. seealso::

986

987 :ref:`orm_queryguide_relationship_per_entity_wildcard` - illustrates an

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

989

990 """

991

992 __slots__ = (

993 "path",

994 "context",

995 "additional_source_entities",

996 )

997

998 _traverse_internals = [

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

1000 (

1001 "context",

1002 visitors.InternalTraversal.dp_has_cache_key_list,

1003 ),

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

1005 (

1006 "additional_source_entities",

1007 visitors.InternalTraversal.dp_has_cache_key_list,

1008 ),

1009 ]

1010 _cache_key_traversal = None

1011

1012 path: PathRegistry

1013 context: Tuple[_LoadElement, ...]

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

1015

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

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

1018 insp._post_inspect

1019

1020 self.path = insp._path_registry

1021 self.context = ()

1022 self.propagate_to_loaders = False

1023 self.additional_source_entities = ()

1024

1025 def __str__(self) -> str:

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

1027

1028 @classmethod

1029 def _construct_for_existing_path(

1030 cls, path: _AbstractEntityRegistry

1031 ) -> Load:

1032 load = cls.__new__(cls)

1033 load.path = path

1034 load.context = ()

1035 load.propagate_to_loaders = False

1036 load.additional_source_entities = ()

1037 return load

1038

1039 def _adapt_cached_option_to_uncached_option(

1040 self, context: QueryContext, uncached_opt: ORMOption

1041 ) -> ORMOption:

1042 if uncached_opt is self:

1043 return self

1044 return self._adjust_for_extra_criteria(context)

1045

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

1047 cloned = self._clone()

1048 cloned.context = tuple(

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

1050 )

1051 return cloned

1052

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

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

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

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

1057

1058 """

1059

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

1061 # actually have any extra_criteria options, which is the

1062 # common case

1063 for value in self.context:

1064 if value._extra_criteria:

1065 break

1066 else:

1067 return self

1068

1069 replacement_cache_key = context.user_passed_query._generate_cache_key()

1070

1071 if replacement_cache_key is None:

1072 return self

1073

1074 orig_query = context.compile_state.select_statement

1075 orig_cache_key = orig_query._generate_cache_key()

1076 assert orig_cache_key is not None

1077

1078 def process(

1079 opt: _LoadElement,

1080 replacement_cache_key: CacheKey,

1081 orig_cache_key: CacheKey,

1082 ) -> _LoadElement:

1083 cloned_opt = opt._clone()

1084

1085 cloned_opt._extra_criteria = tuple(

1086 replacement_cache_key._apply_params_to_element(

1087 orig_cache_key, crit

1088 )

1089 for crit in cloned_opt._extra_criteria

1090 )

1091

1092 return cloned_opt

1093

1094 cloned = self._clone()

1095 cloned.context = tuple(

1096 (

1097 process(value, replacement_cache_key, orig_cache_key)

1098 if value._extra_criteria

1099 else value

1100 )

1101 for value in self.context

1102 )

1103 return cloned

1104

1105 def _reconcile_query_entities_with_us(self, mapper_entities, raiseerr):

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

1107 entity inside of _LoadElement objects.

1108

1109 """

1110 path = self.path

1111

1112 for ent in mapper_entities:

1113 ezero = ent.entity_zero

1114 if ezero and orm_util._entity_corresponds_to(

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

1116 # safe to pass to _entity_corresponds_to()

1117 ezero,

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

1119 ):

1120 return ezero

1121

1122 return None

1123

1124 def _process(

1125 self,

1126 compile_state: _ORMCompileState,

1127 mapper_entities: Sequence[_MapperEntity],

1128 raiseerr: bool,

1129 ) -> None:

1130 reconciled_lead_entity = self._reconcile_query_entities_with_us(

1131 mapper_entities, raiseerr

1132 )

1133

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

1135 has_current_path = bool(compile_state.compile_options._current_path)

1136

1137 for loader in self.context:

1138 # issue #11292

1139 # historically, propagate_to_loaders was only considered at

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

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

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

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

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

1145 # so we check again

1146 if has_current_path and not loader.propagate_to_loaders:

1147 continue

1148 loader.process_compile_state(

1149 self,

1150 compile_state,

1151 mapper_entities,

1152 reconciled_lead_entity,

1153 raiseerr,

1154 )

1155

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

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

1158 :class:`_orm.Load` object.

1159

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

1161

1162 """

1163 cloned = self._generate()

1164

1165 assert cloned.propagate_to_loaders == self.propagate_to_loaders

1166

1167 if not any(

1168 orm_util._entity_corresponds_to_use_path_impl(

1169 elem, cloned.path.odd_element(0)

1170 )

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

1172 + parent.additional_source_entities

1173 ):

1174 if len(cloned.path) > 1:

1175 attrname = cloned.path[1]

1176 parent_entity = cloned.path[0]

1177 else:

1178 attrname = cloned.path[0]

1179 parent_entity = cloned.path[0]

1180 _raise_for_does_not_link(parent.path, attrname, parent_entity)

1181

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

1183

1184 if self.context:

1185 cloned.context = tuple(

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

1187 )

1188

1189 if cloned.context:

1190 parent.context += cloned.context

1191 parent.additional_source_entities += (

1192 cloned.additional_source_entities

1193 )

1194

1195 @_generative

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

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

1198 :class:`_orm.Load`

1199 object.

1200

1201 E.g.::

1202

1203 query = session.query(Author)

1204 query = query.options(

1205 joinedload(Author.book).options(

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

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

1208 )

1209 )

1210

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

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

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

1214

1215 .. seealso::

1216

1217 :func:`.defaultload`

1218

1219 :ref:`orm_queryguide_relationship_sub_options`

1220

1221 """

1222 for opt in opts:

1223 try:

1224 opt._apply_to_parent(self)

1225 except AttributeError as ae:

1226 if not isinstance(opt, _AbstractLoad):

1227 raise sa_exc.ArgumentError(

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

1229 "Load.options() method."

1230 ) from ae

1231 else:

1232 raise

1233 return self

1234

1235 def _clone_for_bind_strategy(

1236 self,

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

1238 strategy: Optional[_StrategyKey],

1239 wildcard_key: Optional[_WildcardKeyType],

1240 opts: Optional[_OptsType] = None,

1241 attr_group: Optional[_AttrGroupType] = None,

1242 propagate_to_loaders: bool = True,

1243 reconcile_to_other: Optional[bool] = None,

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

1245 ) -> Self:

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

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

1248 # InstanceState.load_options

1249 if propagate_to_loaders:

1250 self.propagate_to_loaders = True

1251

1252 if self.path.is_token:

1253 raise sa_exc.ArgumentError(

1254 "Wildcard token cannot be followed by another entity"

1255 )

1256

1257 elif path_is_property(self.path):

1258 # reuse the lookup which will raise a nicely formatted

1259 # LoaderStrategyException

1260 if strategy:

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

1262 else:

1263 raise sa_exc.ArgumentError(

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

1265 "refer to a mapped entity"

1266 )

1267

1268 if attrs is None:

1269 load_element = _ClassStrategyLoad.create(

1270 self.path,

1271 None,

1272 strategy,

1273 wildcard_key,

1274 opts,

1275 propagate_to_loaders,

1276 attr_group=attr_group,

1277 reconcile_to_other=reconcile_to_other,

1278 extra_criteria=extra_criteria,

1279 )

1280 if load_element:

1281 self.context += (load_element,)

1282 assert opts is not None

1283 self.additional_source_entities += cast(

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

1285 )

1286

1287 else:

1288 for attr in attrs:

1289 if isinstance(attr, str):

1290 load_element = _TokenStrategyLoad.create(

1291 self.path,

1292 attr,

1293 strategy,

1294 wildcard_key,

1295 opts,

1296 propagate_to_loaders,

1297 attr_group=attr_group,

1298 reconcile_to_other=reconcile_to_other,

1299 extra_criteria=extra_criteria,

1300 )

1301 else:

1302 load_element = _AttributeStrategyLoad.create(

1303 self.path,

1304 attr,

1305 strategy,

1306 wildcard_key,

1307 opts,

1308 propagate_to_loaders,

1309 attr_group=attr_group,

1310 reconcile_to_other=reconcile_to_other,

1311 extra_criteria=extra_criteria,

1312 )

1313

1314 if load_element:

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

1316 # object with the latest path.

1317 if wildcard_key is _RELATIONSHIP_TOKEN:

1318 self.path = load_element.path

1319 self.context += (load_element,)

1320

1321 # this seems to be effective for selectinloader,

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

1323 # but does not work for immediateloader, which still

1324 # must add additional options at load time

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

1326 r1 = load_element._recurse()

1327 self.context += (r1,)

1328

1329 return self

1330

1331 def __getstate__(self):

1332 d = self._shallow_to_dict()

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

1334 return d

1335

1336 def __setstate__(self, state):

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

1338 self._shallow_from_dict(state)

1339

1340

1341class _WildcardLoad(_AbstractLoad):

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

1343

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

1345

1346 _traverse_internals = [

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

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

1349 (

1350 "local_opts",

1351 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1352 ),

1353 ]

1354 cache_key_traversal: _CacheKeyTraversalType = None

1355

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

1357 local_opts: _OptsType

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

1359 propagate_to_loaders = False

1360

1361 def __init__(self) -> None:

1362 self.path = ()

1363 self.strategy = None

1364 self.local_opts = util.EMPTY_DICT

1365

1366 def _clone_for_bind_strategy(

1367 self,

1368 attrs,

1369 strategy,

1370 wildcard_key,

1371 opts=None,

1372 attr_group=None,

1373 propagate_to_loaders=True,

1374 reconcile_to_other=None,

1375 extra_criteria=None,

1376 ):

1377 assert attrs is not None

1378 attr = attrs[0]

1379 assert (

1380 wildcard_key

1381 and isinstance(attr, str)

1382 and attr in (_WILDCARD_TOKEN, _DEFAULT_TOKEN)

1383 )

1384

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

1386

1387 self.strategy = strategy

1388 self.path = (attr,)

1389 if opts:

1390 self.local_opts = util.immutabledict(opts)

1391

1392 assert extra_criteria is None

1393

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

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

1396

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

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

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

1400

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

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

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

1404

1405 """

1406 assert self.path

1407 attr = self.path[0]

1408 if attr.endswith(_DEFAULT_TOKEN):

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

1410

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

1412

1413 assert effective_path.is_token

1414

1415 loader = _TokenStrategyLoad.create(

1416 effective_path,

1417 None,

1418 self.strategy,

1419 None,

1420 self.local_opts,

1421 self.propagate_to_loaders,

1422 )

1423

1424 parent.context += (loader,)

1425

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

1427 is_refresh = compile_state.compile_options._for_refresh_state

1428

1429 if is_refresh and not self.propagate_to_loaders:

1430 return

1431

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

1433 current_path = compile_state.current_path

1434

1435 start_path: _PathRepresentation = self.path

1436

1437 if current_path:

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

1439 # None back here

1440 new_path = self._chop_path(start_path, current_path)

1441 if new_path is None:

1442 return

1443

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

1445 # just returns it

1446 assert new_path == start_path

1447

1448 # start_path is a single-token tuple

1449 assert start_path and len(start_path) == 1

1450

1451 token = start_path[0]

1452 assert isinstance(token, str)

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

1454

1455 if not entity:

1456 return

1457

1458 path_element = entity

1459

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

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

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

1463 # tokens and populate into the Load().

1464

1465 assert isinstance(token, str)

1466 loader = _TokenStrategyLoad.create(

1467 path_element._path_registry,

1468 token,

1469 self.strategy,

1470 None,

1471 self.local_opts,

1472 self.propagate_to_loaders,

1473 raiseerr=raiseerr,

1474 )

1475 if not loader:

1476 return

1477

1478 assert loader.path.is_token

1479

1480 # don't pass a reconciled lead entity here

1481 loader.process_compile_state(

1482 self, compile_state, mapper_entities, None, raiseerr

1483 )

1484

1485 return loader

1486

1487 def _find_entity_basestring(

1488 self,

1489 entities: Iterable[_InternalEntityType[Any]],

1490 token: str,

1491 raiseerr: bool,

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

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

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

1495 if raiseerr:

1496 raise sa_exc.ArgumentError(

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

1498 f"loader option to multiple entities "

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

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

1501 f"""{

1502 ", ".join(

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

1504 for ent in entities

1505 )

1506 }."""

1507 )

1508 elif token.endswith(_DEFAULT_TOKEN):

1509 raiseerr = False

1510

1511 for ent in entities:

1512 # return only the first _MapperEntity when searching

1513 # based on string prop name. Ideally object

1514 # attributes are used to specify more exactly.

1515 return ent

1516 else:

1517 if raiseerr:

1518 raise sa_exc.ArgumentError(

1519 "Query has only expression-based entities - "

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

1521 )

1522 else:

1523 return None

1524

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

1526 d = self._shallow_to_dict()

1527 return d

1528

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

1530 self._shallow_from_dict(state)

1531

1532

1533class _LoadElement(

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

1535):

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

1537 and pass options to it.

1538

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

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

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

1542

1543 .. versionadded:: 2.0

1544

1545 """

1546

1547 __slots__ = (

1548 "path",

1549 "strategy",

1550 "propagate_to_loaders",

1551 "local_opts",

1552 "_extra_criteria",

1553 "_reconcile_to_other",

1554 )

1555 __visit_name__ = "load_element"

1556

1557 _traverse_internals = [

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

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

1560 (

1561 "local_opts",

1562 visitors.ExtendedInternalTraversal.dp_string_multi_dict,

1563 ),

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

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

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

1567 ]

1568 _cache_key_traversal = None

1569

1570 _extra_criteria: Tuple[Any, ...]

1571

1572 _reconcile_to_other: Optional[bool]

1573 strategy: Optional[_StrategyKey]

1574 path: PathRegistry

1575 propagate_to_loaders: bool

1576

1577 local_opts: util.immutabledict[str, Any]

1578

1579 is_token_strategy: bool

1580 is_class_strategy: bool

1581

1582 def __hash__(self) -> int:

1583 return id(self)

1584

1585 def __eq__(self, other):

1586 return traversals.compare(self, other)

1587

1588 @property

1589 def is_opts_only(self) -> bool:

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

1591

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

1593 cls = self.__class__

1594 s = cls.__new__(cls)

1595

1596 self._shallow_copy_to(s)

1597 return s

1598

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

1600 new = self._clone()

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

1602 return new

1603

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

1605 d = self._shallow_to_dict()

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

1607 return d

1608

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

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

1611 self._shallow_from_dict(state)

1612

1613 def _raise_for_no_match(self, parent_loader, mapper_entities):

1614 path = parent_loader.path

1615

1616 found_entities = False

1617 for ent in mapper_entities:

1618 ezero = ent.entity_zero

1619 if ezero:

1620 found_entities = True

1621 break

1622

1623 if not found_entities:

1624 raise sa_exc.ArgumentError(

1625 "Query has only expression-based entities; "

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

1627 "can't be applied here."

1628 )

1629 else:

1630 raise sa_exc.ArgumentError(

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

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

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

1634 f"""{

1635 ", ".join(

1636 entity_str(x.entity_zero)

1637 for x in mapper_entities if x.entity_zero

1638 )}. Please """

1639 "specify the full path "

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

1641 "attribute. "

1642 )

1643

1644 def _adjust_effective_path_for_current_path(

1645 self, effective_path: PathRegistry, current_path: PathRegistry

1646 ) -> Optional[PathRegistry]:

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

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

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

1650 current_path.

1651

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

1653

1654 .. sourcecode:: text

1655

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

1657

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

1659

1660 The adjusted path would be:

1661

1662 .. sourcecode:: text

1663

1664 Item -> keywords -> Keyword

1665

1666

1667 """

1668 chopped_start_path = Load._chop_path(

1669 effective_path.natural_path, current_path

1670 )

1671 if not chopped_start_path:

1672 return None

1673

1674 tokens_removed_from_start_path = len(effective_path) - len(

1675 chopped_start_path

1676 )

1677

1678 loader_lead_path_element = self.path[tokens_removed_from_start_path]

1679

1680 effective_path = PathRegistry.coerce(

1681 (loader_lead_path_element,) + chopped_start_path[1:]

1682 )

1683

1684 return effective_path

1685

1686 def _init_path(

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

1688 ):

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

1690 a new path.

1691

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

1693 a :class:`._LoadElement` object.

1694

1695 """

1696 raise NotImplementedError()

1697

1698 def _prepare_for_compile_state(

1699 self,

1700 parent_loader,

1701 compile_state,

1702 mapper_entities,

1703 reconciled_lead_entity,

1704 raiseerr,

1705 ):

1706 """implemented by subclasses."""

1707 raise NotImplementedError()

1708

1709 def process_compile_state(

1710 self,

1711 parent_loader,

1712 compile_state,

1713 mapper_entities,

1714 reconciled_lead_entity,

1715 raiseerr,

1716 ):

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

1718 _LoadElement.

1719

1720 """

1721 keys = self._prepare_for_compile_state(

1722 parent_loader,

1723 compile_state,

1724 mapper_entities,

1725 reconciled_lead_entity,

1726 raiseerr,

1727 )

1728 for key in keys:

1729 if key in compile_state.attributes:

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

1731 self, compile_state.attributes[key]

1732 )

1733 else:

1734 compile_state.attributes[key] = self

1735

1736 @classmethod

1737 def create(

1738 cls,

1739 path: PathRegistry,

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

1741 strategy: Optional[_StrategyKey],

1742 wildcard_key: Optional[_WildcardKeyType],

1743 local_opts: Optional[_OptsType],

1744 propagate_to_loaders: bool,

1745 raiseerr: bool = True,

1746 attr_group: Optional[_AttrGroupType] = None,

1747 reconcile_to_other: Optional[bool] = None,

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

1749 ) -> _LoadElement:

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

1751

1752 opt = cls.__new__(cls)

1753 opt.path = path

1754 opt.strategy = strategy

1755 opt.propagate_to_loaders = propagate_to_loaders

1756 opt.local_opts = (

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

1758 )

1759 opt._extra_criteria = ()

1760

1761 if reconcile_to_other is not None:

1762 opt._reconcile_to_other = reconcile_to_other

1763 elif strategy is None and not local_opts:

1764 opt._reconcile_to_other = True

1765 else:

1766 opt._reconcile_to_other = None

1767

1768 path = opt._init_path(

1769 path, attr, wildcard_key, attr_group, raiseerr, extra_criteria

1770 )

1771

1772 if not path:

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

1774

1775 assert opt.is_token_strategy == path.is_token

1776

1777 opt.path = path

1778 return opt

1779

1780 def __init__(self) -> None:

1781 raise NotImplementedError()

1782

1783 def _recurse(self) -> _LoadElement:

1784 cloned = self._clone()

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

1786

1787 return cloned

1788

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

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

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

1792 path.

1793

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

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

1796

1797 """

1798

1799 if not any(

1800 orm_util._entity_corresponds_to_use_path_impl(

1801 elem,

1802 self.path.odd_element(0),

1803 )

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

1805 + parent.additional_source_entities

1806 ):

1807 raise sa_exc.ArgumentError(

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

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

1810 )

1811

1812 return self._prepend_path(parent.path)

1813

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

1815 cloned = self._clone()

1816

1817 assert cloned.strategy == self.strategy

1818 assert cloned.local_opts == self.local_opts

1819 assert cloned.is_class_strategy == self.is_class_strategy

1820

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

1822

1823 return cloned

1824

1825 @staticmethod

1826 def _reconcile(

1827 replacement: _LoadElement, existing: _LoadElement

1828 ) -> _LoadElement:

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

1830 the context.attributes under the same key.

1831

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

1833 existing one

1834

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

1836

1837 """

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

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

1840 # see test_poly_loading.py

1841

1842 if replacement._reconcile_to_other:

1843 return existing

1844 elif replacement._reconcile_to_other is False:

1845 return replacement

1846 elif existing._reconcile_to_other:

1847 return replacement

1848 elif existing._reconcile_to_other is False:

1849 return existing

1850

1851 if existing is replacement:

1852 return replacement

1853 elif (

1854 existing.strategy == replacement.strategy

1855 and existing.local_opts == replacement.local_opts

1856 ):

1857 return replacement

1858 elif replacement.is_opts_only:

1859 existing = existing._clone()

1860 existing.local_opts = existing.local_opts.union(

1861 replacement.local_opts

1862 )

1863 existing._extra_criteria += replacement._extra_criteria

1864 return existing

1865 elif existing.is_opts_only:

1866 replacement = replacement._clone()

1867 replacement.local_opts = replacement.local_opts.union(

1868 existing.local_opts

1869 )

1870 replacement._extra_criteria += existing._extra_criteria

1871 return replacement

1872 elif replacement.path.is_token:

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

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

1875 # will raise as below

1876 return replacement

1877

1878 raise sa_exc.InvalidRequestError(

1879 f"Loader strategy replacement "

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

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

1882 )

1883

1884 def _to_option_method_string(self) -> str:

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

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

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

1888

1889 """

1890 assert (

1891 self.strategy is not None

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

1893

1894 for opt_key in self.local_opts:

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

1896 if fn:

1897 break

1898 else:

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

1900

1901 assert (

1902 fn is not None

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

1904

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

1906

1907

1908class _AttributeStrategyLoad(_LoadElement):

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

1910

1911 e.g.::

1912

1913 joinedload(User.addresses)

1914 defer(Order.name)

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

1916

1917 """

1918

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

1920

1921 __visit_name__ = "attribute_strategy_load_element"

1922

1923 _traverse_internals = _LoadElement._traverse_internals + [

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

1925 (

1926 "_path_with_polymorphic_path",

1927 visitors.ExtendedInternalTraversal.dp_has_cache_key,

1928 ),

1929 ]

1930

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

1932 _path_with_polymorphic_path: Optional[PathRegistry]

1933

1934 is_class_strategy = False

1935 is_token_strategy = False

1936

1937 def _init_path(

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

1939 ):

1940 assert attr is not None

1941 self._of_type = None

1942 self._path_with_polymorphic_path = None

1943 insp, _, prop = _parse_attr_argument(attr)

1944

1945 if insp.is_property:

1946 # direct property can be sent from internal strategy logic

1947 # that sets up specific loaders, such as

1948 # emit_lazyload->_lazyload_reverse

1949 # prop = found_property = attr

1950 prop = attr

1951 path = path[prop]

1952

1953 if path.has_entity:

1954 path = path.entity_path

1955 return path

1956

1957 elif not insp.is_attribute:

1958 # should not reach here;

1959 assert False

1960

1961 # here we assume we have user-passed InstrumentedAttribute

1962 if not orm_util._entity_corresponds_to_use_path_impl(

1963 path[-1], attr.parent

1964 ):

1965 if raiseerr:

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

1967 raise sa_exc.ArgumentError(

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

1969 "loader option to multiple entities in the "

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

1971 )

1972 else:

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

1974 else:

1975 return None

1976

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

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

1979 # test_relationship_criteria.py::RelationshipCriteriaTest::

1980 # test_selectinload_nested_criteria[True] if an existing

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

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

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

1984 # poorly defined.

1985 if extra_criteria:

1986 assert not attr._extra_criteria

1987 self._extra_criteria = extra_criteria

1988 else:

1989 self._extra_criteria = attr._extra_criteria

1990

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

1992 ac = attr._of_type

1993 ext_info = inspect(ac)

1994 self._of_type = ext_info

1995

1996 self._path_with_polymorphic_path = path.entity_path[prop]

1997

1998 path = path[prop][ext_info]

1999

2000 else:

2001 path = path[prop]

2002

2003 if path.has_entity:

2004 path = path.entity_path

2005

2006 return path

2007

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

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

2010

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

2012 ensures that the polymorphic path information is correctly updated

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

2014 """

2015 cloned = super()._prepend_path(path)

2016

2017 # Also prepend the parent path to _path_with_polymorphic_path if

2018 # present

2019 if self._path_with_polymorphic_path is not None:

2020 cloned._path_with_polymorphic_path = PathRegistry.coerce(

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

2022 )

2023

2024 return cloned

2025

2026 def _generate_extra_criteria(self, context):

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

2028 immediate "extra_criteria" stored with this Load object.

2029

2030 Load objects are typically pulled from the cached version of

2031 the statement from a QueryContext. The statement currently being

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

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

2034 they handle this criteria for a result set.

2035

2036 """

2037

2038 assert (

2039 self._extra_criteria

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

2041

2042 orig_query = context.compile_state.select_statement

2043 current_query = context.query

2044

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

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

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

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

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

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

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

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

2053

2054 # if orig_query is current_query:

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

2056 # return and_(*self._extra_criteria)

2057

2058 k1 = orig_query._generate_cache_key()

2059 k2 = current_query._generate_cache_key()

2060

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

2062

2063 def _set_of_type_info(self, context, current_path):

2064 assert self._path_with_polymorphic_path

2065

2066 pwpi = self._of_type

2067 assert pwpi

2068 if not pwpi.is_aliased_class:

2069 pwpi = inspect(

2070 orm_util.AliasedInsp._with_polymorphic_factory(

2071 pwpi.mapper.base_mapper,

2072 (pwpi.mapper,),

2073 aliased=True,

2074 _use_mapper_path=True,

2075 )

2076 )

2077 start_path = self._path_with_polymorphic_path

2078 if current_path:

2079 new_path = self._adjust_effective_path_for_current_path(

2080 start_path, current_path

2081 )

2082 if new_path is None:

2083 return

2084 start_path = new_path

2085

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

2087 if key in context:

2088 existing_aliased_insp = context[key]

2089 this_aliased_insp = pwpi

2090 new_aliased_insp = existing_aliased_insp._merge_with(

2091 this_aliased_insp

2092 )

2093 context[key] = new_aliased_insp

2094 else:

2095 context[key] = pwpi

2096

2097 def _prepare_for_compile_state(

2098 self,

2099 parent_loader,

2100 compile_state,

2101 mapper_entities,

2102 reconciled_lead_entity,

2103 raiseerr,

2104 ):

2105 # _AttributeStrategyLoad

2106

2107 current_path = compile_state.current_path

2108 is_refresh = compile_state.compile_options._for_refresh_state

2109 assert not self.path.is_token

2110

2111 if is_refresh and not self.propagate_to_loaders:

2112 return []

2113

2114 if self._of_type:

2115 # apply additional with_polymorphic alias that may have been

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

2117 self._set_of_type_info(compile_state.attributes, current_path)

2118

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

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

2121 return []

2122

2123 if raiseerr and not reconciled_lead_entity:

2124 self._raise_for_no_match(parent_loader, mapper_entities)

2125

2126 if self.path.has_entity:

2127 effective_path = self.path.parent

2128 else:

2129 effective_path = self.path

2130

2131 if current_path:

2132 assert effective_path is not None

2133 effective_path = self._adjust_effective_path_for_current_path(

2134 effective_path, current_path

2135 )

2136 if effective_path is None:

2137 return []

2138

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

2140

2141 def __getstate__(self):

2142 d = super().__getstate__()

2143

2144 # can't pickle this. See

2145 # test_pickled.py -> test_lazyload_extra_criteria_not_supported

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

2147 # would be non-None

2148 d["_extra_criteria"] = ()

2149

2150 if self._path_with_polymorphic_path:

2151 d["_path_with_polymorphic_path"] = (

2152 self._path_with_polymorphic_path.serialize()

2153 )

2154

2155 if self._of_type:

2156 if self._of_type.is_aliased_class:

2157 d["_of_type"] = None

2158 elif self._of_type.is_mapper:

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

2160 else:

2161 assert False, "unexpected object for _of_type"

2162

2163 return d

2164

2165 def __setstate__(self, state):

2166 super().__setstate__(state)

2167

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

2169 self._path_with_polymorphic_path = PathRegistry.deserialize(

2170 state["_path_with_polymorphic_path"]

2171 )

2172 else:

2173 self._path_with_polymorphic_path = None

2174

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

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

2177 else:

2178 self._of_type = None

2179

2180

2181class _TokenStrategyLoad(_LoadElement):

2182 """Loader strategies against wildcard attributes

2183

2184 e.g.::

2185

2186 raiseload("*")

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

2188 defer("*")

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

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

2191

2192 """

2193

2194 __visit_name__ = "token_strategy_load_element"

2195

2196 inherit_cache = True

2197 is_class_strategy = False

2198 is_token_strategy = True

2199

2200 def _init_path(

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

2202 ):

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

2204 if attr is not None:

2205 default_token = attr.endswith(_DEFAULT_TOKEN)

2206 if attr.endswith(_WILDCARD_TOKEN) or default_token:

2207 if wildcard_key:

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

2209

2210 path = path.token(attr)

2211 return path

2212 else:

2213 raise sa_exc.ArgumentError(

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

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

2216 )

2217 return path

2218

2219 def _prepare_for_compile_state(

2220 self,

2221 parent_loader,

2222 compile_state,

2223 mapper_entities,

2224 reconciled_lead_entity,

2225 raiseerr,

2226 ):

2227 # _TokenStrategyLoad

2228

2229 current_path = compile_state.current_path

2230 is_refresh = compile_state.compile_options._for_refresh_state

2231

2232 assert self.path.is_token

2233

2234 if is_refresh and not self.propagate_to_loaders:

2235 return []

2236

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

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

2239 return []

2240

2241 effective_path = self.path

2242 if reconciled_lead_entity:

2243 effective_path = PathRegistry.coerce(

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

2245 )

2246

2247 if current_path:

2248 new_effective_path = self._adjust_effective_path_for_current_path(

2249 effective_path, current_path

2250 )

2251 if new_effective_path is None:

2252 return []

2253 effective_path = new_effective_path

2254

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

2256 # to encompass everything from the query entity on

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

2258 # is set.

2259

2260 return [

2261 ("loader", natural_path)

2262 for natural_path in (

2263 cast(

2264 _TokenRegistry, effective_path

2265 )._generate_natural_for_superclasses()

2266 )

2267 ]

2268

2269

2270class _ClassStrategyLoad(_LoadElement):

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

2272 an attribute path

2273

2274 e.g.::

2275

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

2277 selectin_polymorphic(Person, [Engineer, Manager])

2278 )

2279

2280 """

2281

2282 inherit_cache = True

2283 is_class_strategy = True

2284 is_token_strategy = False

2285

2286 __visit_name__ = "class_strategy_load_element"

2287

2288 def _init_path(

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

2290 ):

2291 return path

2292

2293 def _prepare_for_compile_state(

2294 self,

2295 parent_loader,

2296 compile_state,

2297 mapper_entities,

2298 reconciled_lead_entity,

2299 raiseerr,

2300 ):

2301 # _ClassStrategyLoad

2302

2303 current_path = compile_state.current_path

2304 is_refresh = compile_state.compile_options._for_refresh_state

2305

2306 if is_refresh and not self.propagate_to_loaders:

2307 return []

2308

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

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

2311 return []

2312

2313 effective_path = self.path

2314

2315 if current_path:

2316 new_effective_path = self._adjust_effective_path_for_current_path(

2317 effective_path, current_path

2318 )

2319 if new_effective_path is None:

2320 return []

2321 effective_path = new_effective_path

2322

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

2324

2325

2326def _generate_from_keys(

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

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

2329 chained: bool,

2330 kw: Any,

2331) -> _AbstractLoad:

2332 lead_element: Optional[_AbstractLoad] = None

2333

2334 attr: Any

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

2336 for attr in _keys:

2337 if isinstance(attr, str):

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

2339 util.warn_deprecated(

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

2341 "deprecated "

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

2343 "it is "

2344 "believed to be unused. "

2345 "If you have been using this functionality, "

2346 "please "

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

2348 "tracker.",

2349 version="1.4",

2350 )

2351 attr = attr[1:]

2352

2353 if attr == _WILDCARD_TOKEN:

2354 if is_default:

2355 raise sa_exc.ArgumentError(

2356 "Wildcard token cannot be followed by "

2357 "another entity",

2358 )

2359

2360 if lead_element is None:

2361 lead_element = _WildcardLoad()

2362

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

2364

2365 else:

2366 raise sa_exc.ArgumentError(

2367 "Strings are not accepted for attribute names in "

2368 "loader options; please use class-bound "

2369 "attributes directly.",

2370 )

2371 else:

2372 if lead_element is None:

2373 _, lead_entity, _ = _parse_attr_argument(attr)

2374 lead_element = Load(lead_entity)

2375

2376 if is_default:

2377 if not chained:

2378 lead_element = lead_element.defaultload(attr)

2379 else:

2380 lead_element = meth(

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

2382 )

2383 else:

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

2385

2386 assert lead_element

2387 return lead_element

2388

2389

2390def _parse_attr_argument(

2391 attr: _AttrType,

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

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

2394 :class:`._AbstractLoad` instance.

2395

2396 This is used by the standalone loader strategy functions like

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

2398 :class:`._WildcardLoad` objects.

2399

2400 """

2401 try:

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

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

2404 # if at all

2405 insp: InspectionAttr = inspect(attr) # type: ignore[assignment]

2406 except sa_exc.NoInspectionAvailable as err:

2407 raise sa_exc.ArgumentError(

2408 "expected ORM mapped attribute for loader strategy argument"

2409 ) from err

2410

2411 lead_entity: _InternalEntityType[Any]

2412

2413 if insp_is_mapper_property(insp):

2414 lead_entity = insp.parent

2415 prop = insp

2416 elif insp_is_attribute(insp):

2417 lead_entity = insp.parent

2418 prop = insp.prop

2419 else:

2420 raise sa_exc.ArgumentError(

2421 "expected ORM mapped attribute for loader strategy argument"

2422 )

2423

2424 return insp, lead_entity, prop

2425

2426

2427def _strategy_labels(

2428 *strategy_keys: "Any",

2429 discriminating_opt: Optional[str] = None,

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

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

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

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

2434 runs.

2435

2436 :param discriminating_opt: optional local_opts key that distinguishes

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

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

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

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

2441 """

2442

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

2444 for key in strategy_keys:

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

2446 return fn

2447

2448 return decorator

2449

2450

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

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

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

2454

2455 """

2456 bound_fn = getattr(_AbstractLoad, fn.__name__)

2457 fn_doc = bound_fn.__doc__

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

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

2460

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

2462

2463"""

2464

2465 fn.__doc__ = fn_doc

2466 return fn

2467

2468

2469def _expand_column_strategy_attrs(

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

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

2472 return cast(

2473 "Tuple[_AttrType, ...]",

2474 tuple(

2475 a

2476 for attr in attrs

2477 for a in (

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

2479 if hasattr(attr, "_column_strategy_attrs")

2480 else (attr,)

2481 )

2482 ),

2483 )

2484

2485

2486# standalone functions follow. docstrings are filled in

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

2488

2489

2490@loader_unbound_fn

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

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

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

2494

2495

2496@loader_unbound_fn

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

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

2499 # add some extra state to Load of some kind

2500 attrs = _expand_column_strategy_attrs(attrs)

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

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

2503

2504

2505@loader_unbound_fn

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

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

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

2509

2510

2511@loader_unbound_fn

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

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

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

2515

2516

2517@loader_unbound_fn

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

2519def selectinload(

2520 *keys: _AttrType,

2521 recursion_depth: Optional[int] = None,

2522 chunksize: Optional[int] = None,

2523) -> _AbstractLoad:

2524 return _generate_from_keys(

2525 Load.selectinload,

2526 keys,

2527 False,

2528 {"recursion_depth": recursion_depth, "chunksize": chunksize},

2529 )

2530

2531

2532@loader_unbound_fn

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

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

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

2536

2537

2538@loader_unbound_fn

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

2540def immediateload(

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

2542) -> _AbstractLoad:

2543 return _generate_from_keys(

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

2545 )

2546

2547

2548@loader_unbound_fn

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

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

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

2552

2553

2554@loader_unbound_fn

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

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

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

2558

2559

2560@loader_unbound_fn

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

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

2563

2564

2565@loader_unbound_fn

2566@_strategy_labels(

2567 (("deferred", True), ("instrument", True)),

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

2569)

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

2571 if raiseload:

2572 kw = {"raiseload": raiseload}

2573 else:

2574 kw = {}

2575

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

2577

2578

2579@loader_unbound_fn

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

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

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

2583

2584

2585@loader_unbound_fn

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

2587 element = _WildcardLoad()

2588 return element.undefer_group(name)

2589

2590

2591@loader_unbound_fn

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

2593def with_expression(

2594 key: _AttrType, expression: _ColumnExpressionArgument[Any]

2595) -> _AbstractLoad:

2596 return _generate_from_keys(

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

2598 )

2599

2600

2601@loader_unbound_fn

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

2603def selectin_polymorphic(

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

2605) -> _AbstractLoad:

2606 ul = Load(base_cls)

2607 return ul.selectin_polymorphic(classes)

2608

2609

2610def _raise_for_does_not_link(path, attrname, parent_entity):

2611 if len(path) > 1:

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

2613

2614 raise sa_exc.ArgumentError(

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

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

2617 % (

2618 (

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

2620 if path_is_of_type

2621 else ""

2622 ),

2623 (

2624 " Did you mean to use "

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

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

2627 '"loadopt.options('

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

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

2630 if not path_is_of_type

2631 and not path[-1].is_aliased_class

2632 and orm_util._entity_corresponds_to(

2633 path.entity, inspect(parent_entity).mapper

2634 )

2635 else ""

2636 ),

2637 )

2638 )

2639 else:

2640 raise sa_exc.ArgumentError(

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

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

2643 )

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

813 statements