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

1# orm/strategies.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

8"""sqlalchemy.orm.interfaces.LoaderStrategy

9 implementations, and related MapperOptions."""

10from __future__ import absolute_import

12import collections

13import itertools

15from . import attributes

16from . import exc as orm_exc

17from . import interfaces

18from . import loading

19from . import path_registry

20from . import properties

21from . import query

22from . import relationships

23from . import unitofwork

24from . import util as orm_util

25from .base import _DEFER_FOR_STATE

26from .base import _RAISE_FOR_STATE

27from .base import _SET_DEFERRED_EXPIRED

28from .context import _column_descriptions

29from .context import ORMCompileState

30from .context import ORMSelectCompileState

31from .context import QueryContext

32from .interfaces import LoaderStrategy

33from .interfaces import StrategizedProperty

34from .session import _state_session

35from .state import InstanceState

36from .util import _none_set

37from .util import aliased

38from .. import event

39from .. import exc as sa_exc

40from .. import inspect

41from .. import log

42from .. import sql

43from .. import util

44from ..sql import util as sql_util

45from ..sql import visitors

46from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL

47from ..sql.selectable import Select

50def _register_attribute(

51 prop,

52 mapper,

53 useobject,

54 compare_function=None,

55 typecallable=None,

56 callable_=None,

57 proxy_property=None,

58 active_history=False,

59 impl_class=None,

60 **kw

61):

63 listen_hooks = []

65 uselist = useobject and prop.uselist

67 if useobject and prop.single_parent:

68 listen_hooks.append(single_parent_validator)

70 if prop.key in prop.parent.validators:

71 fn, opts = prop.parent.validators[prop.key]

72 listen_hooks.append(

73 lambda desc, prop: orm_util._validator_events(

74 desc, prop.key, fn, **opts

75 )

76 )

78 if useobject:

79 listen_hooks.append(unitofwork.track_cascade_events)

81 # need to assemble backref listeners

82 # after the singleparentvalidator, mapper validator

83 if useobject:

84 backref = prop.back_populates

85 if backref and prop._effective_sync_backref:

86 listen_hooks.append(

87 lambda desc, prop: attributes.backref_listeners(

88 desc, backref, uselist

89 )

90 )

92 # a single MapperProperty is shared down a class inheritance

93 # hierarchy, so we set up attribute instrumentation and backref event

94 # for each mapper down the hierarchy.

96 # typically, "mapper" is the same as prop.parent, due to the way

97 # the configure_mappers() process runs, however this is not strongly

98 # enforced, and in the case of a second configure_mappers() run the

99 # mapper here might not be prop.parent; also, a subclass mapper may

100 # be called here before a superclass mapper. That is, can't depend

101 # on mappers not already being set up so we have to check each one.

102

103 for m in mapper.self_and_descendants:

104 if prop is m._props.get(

105 prop.key

106 ) and not m.class_manager._attr_has_impl(prop.key):

107

108 desc = attributes.register_attribute_impl(

109 m.class_,

110 prop.key,

111 parent_token=prop,

112 uselist=uselist,

113 compare_function=compare_function,

114 useobject=useobject,

115 trackparent=useobject

116 and (

117 prop.single_parent

118 or prop.direction is interfaces.ONETOMANY

119 ),

120 typecallable=typecallable,

121 callable_=callable_,

122 active_history=active_history,

123 impl_class=impl_class,

124 send_modified_events=not useobject or not prop.viewonly,

125 doc=prop.doc,

126 **kw

127 )

128

129 for hook in listen_hooks:

130 hook(desc, prop)

131

132

133@properties.ColumnProperty.strategy_for(instrument=False, deferred=False)

134class UninstrumentedColumnLoader(LoaderStrategy):

135 """Represent a non-instrumented MapperProperty.

136

137 The polymorphic_on argument of mapper() often results in this,

138 if the argument is against the with_polymorphic selectable.

139

140 """

141

142 __slots__ = ("columns",)

143

144 def __init__(self, parent, strategy_key):

145 super(UninstrumentedColumnLoader, self).__init__(parent, strategy_key)

146 self.columns = self.parent_property.columns

147

148 def setup_query(

149 self,

150 compile_state,

151 query_entity,

152 path,

153 loadopt,

154 adapter,

155 column_collection=None,

156 **kwargs

157 ):

158 for c in self.columns:

159 if adapter:

160 c = adapter.columns[c]

161 compile_state._append_dedupe_col_collection(c, column_collection)

162

163 def create_row_processor(

164 self,

165 context,

166 query_entity,

167 path,

168 loadopt,

169 mapper,

170 result,

171 adapter,

172 populators,

173 ):

174 pass

175

176

177@log.class_logger

178@properties.ColumnProperty.strategy_for(instrument=True, deferred=False)

179class ColumnLoader(LoaderStrategy):

180 """Provide loading behavior for a :class:`.ColumnProperty`."""

181

182 __slots__ = "columns", "is_composite"

183

184 def __init__(self, parent, strategy_key):

185 super(ColumnLoader, self).__init__(parent, strategy_key)

186 self.columns = self.parent_property.columns

187 self.is_composite = hasattr(self.parent_property, "composite_class")

188

189 def setup_query(

190 self,

191 compile_state,

192 query_entity,

193 path,

194 loadopt,

195 adapter,

196 column_collection,

197 memoized_populators,

198 check_for_adapt=False,

199 **kwargs

200 ):

201 for c in self.columns:

202 if adapter:

203 if check_for_adapt:

204 c = adapter.adapt_check_present(c)

205 if c is None:

206 return

207 else:

208 c = adapter.columns[c]

209

210 compile_state._append_dedupe_col_collection(c, column_collection)

211

212 fetch = self.columns[0]

213 if adapter:

214 fetch = adapter.columns[fetch]

215 memoized_populators[self.parent_property] = fetch

216

217 def init_class_attribute(self, mapper):

218 self.is_class_level = True

219 coltype = self.columns[0].type

220 # TODO: check all columns ? check for foreign key as well?

221 active_history = (

222 self.parent_property.active_history

223 or self.columns[0].primary_key

224 or (

225 mapper.version_id_col is not None

226 and mapper._columntoproperty.get(mapper.version_id_col, None)

227 is self.parent_property

228 )

229 )

230

231 _register_attribute(

232 self.parent_property,

233 mapper,

234 useobject=False,

235 compare_function=coltype.compare_values,

236 active_history=active_history,

237 )

238

239 def create_row_processor(

240 self,

241 context,

242 query_entity,

243 path,

244 loadopt,

245 mapper,

246 result,

247 adapter,

248 populators,

249 ):

250 # look through list of columns represented here

251 # to see which, if any, is present in the row.

252

253 for col in self.columns:

254 if adapter:

255 col = adapter.columns[col]

256 getter = result._getter(col, False)

257 if getter:

258 populators["quick"].append((self.key, getter))

259 break

260 else:

261 populators["expire"].append((self.key, True))

262

263

264@log.class_logger

265@properties.ColumnProperty.strategy_for(query_expression=True)

266class ExpressionColumnLoader(ColumnLoader):

267 def __init__(self, parent, strategy_key):

268 super(ExpressionColumnLoader, self).__init__(parent, strategy_key)

269

270 # compare to the "default" expression that is mapped in

271 # the column. If it's sql.null, we don't need to render

272 # unless an expr is passed in the options.

273 null = sql.null().label(None)

274 self._have_default_expression = any(

275 not c.compare(null) for c in self.parent_property.columns

276 )

277

278 def setup_query(

279 self,

280 compile_state,

281 query_entity,

282 path,

283 loadopt,

284 adapter,

285 column_collection,

286 memoized_populators,

287 **kwargs

288 ):

289 columns = None

290 if loadopt and "expression" in loadopt.local_opts:

291 columns = [loadopt.local_opts["expression"]]

292 elif self._have_default_expression:

293 columns = self.parent_property.columns

294

295 if columns is None:

296 return

297

298 for c in columns:

299 if adapter:

300 c = adapter.columns[c]

301 compile_state._append_dedupe_col_collection(c, column_collection)

302

303 fetch = columns[0]

304 if adapter:

305 fetch = adapter.columns[fetch]

306 memoized_populators[self.parent_property] = fetch

307

308 def create_row_processor(

309 self,

310 context,

311 query_entity,

312 path,

313 loadopt,

314 mapper,

315 result,

316 adapter,

317 populators,

318 ):

319 # look through list of columns represented here

320 # to see which, if any, is present in the row.

321 if loadopt and "expression" in loadopt.local_opts:

322 columns = [loadopt.local_opts["expression"]]

323

324 for col in columns:

325 if adapter:

326 col = adapter.columns[col]

327 getter = result._getter(col, False)

328 if getter:

329 populators["quick"].append((self.key, getter))

330 break

331 else:

332 populators["expire"].append((self.key, True))

333

334 def init_class_attribute(self, mapper):

335 self.is_class_level = True

336

337 _register_attribute(

338 self.parent_property,

339 mapper,

340 useobject=False,

341 compare_function=self.columns[0].type.compare_values,

342 accepts_scalar_loader=False,

343 )

344

345

346@log.class_logger

347@properties.ColumnProperty.strategy_for(deferred=True, instrument=True)

348@properties.ColumnProperty.strategy_for(

349 deferred=True, instrument=True, raiseload=True

350)

351@properties.ColumnProperty.strategy_for(do_nothing=True)

352class DeferredColumnLoader(LoaderStrategy):

353 """Provide loading behavior for a deferred :class:`.ColumnProperty`."""

354

355 __slots__ = "columns", "group", "raiseload"

356

357 def __init__(self, parent, strategy_key):

358 super(DeferredColumnLoader, self).__init__(parent, strategy_key)

359 if hasattr(self.parent_property, "composite_class"):

360 raise NotImplementedError(

361 "Deferred loading for composite " "types not implemented yet"

362 )

363 self.raiseload = self.strategy_opts.get("raiseload", False)

364 self.columns = self.parent_property.columns

365 self.group = self.parent_property.group

366

367 def create_row_processor(

368 self,

369 context,

370 query_entity,

371 path,

372 loadopt,

373 mapper,

374 result,

375 adapter,

376 populators,

377 ):

378

379 # for a DeferredColumnLoader, this method is only used during a

380 # "row processor only" query; see test_deferred.py ->

381 # tests with "rowproc_only" in their name. As of the 1.0 series,

382 # loading._instance_processor doesn't use a "row processing" function

383 # to populate columns, instead it uses data in the "populators"

384 # dictionary. Normally, the DeferredColumnLoader.setup_query()

385 # sets up that data in the "memoized_populators" dictionary

386 # and "create_row_processor()" here is never invoked.

387

388 if (

389 context.refresh_state

390 and context.query._compile_options._only_load_props

391 and self.key in context.query._compile_options._only_load_props

392 ):

393 self.parent_property._get_strategy(

394 (("deferred", False), ("instrument", True))

395 ).create_row_processor(

396 context,

397 query_entity,

398 path,

399 loadopt,

400 mapper,

401 result,

402 adapter,

403 populators,

404 )

405

406 elif not self.is_class_level:

407 if self.raiseload:

408 set_deferred_for_local_state = (

409 self.parent_property._raise_column_loader

410 )

411 else:

412 set_deferred_for_local_state = (

413 self.parent_property._deferred_column_loader

414 )

415 populators["new"].append((self.key, set_deferred_for_local_state))

416 else:

417 populators["expire"].append((self.key, False))

418

419 def init_class_attribute(self, mapper):

420 self.is_class_level = True

421

422 _register_attribute(

423 self.parent_property,

424 mapper,

425 useobject=False,

426 compare_function=self.columns[0].type.compare_values,

427 callable_=self._load_for_state,

428 load_on_unexpire=False,

429 )

430

431 def setup_query(

432 self,

433 compile_state,

434 query_entity,

435 path,

436 loadopt,

437 adapter,

438 column_collection,

439 memoized_populators,

440 only_load_props=None,

441 **kw

442 ):

443

444 if (

445 (

446 compile_state.compile_options._render_for_subquery

447 and self.parent_property._renders_in_subqueries

448 )

449 or (

450 loadopt

451 and "undefer_pks" in loadopt.local_opts

452 and set(self.columns).intersection(

453 self.parent._should_undefer_in_wildcard

454 )

455 )

456 or (

457 loadopt

458 and self.group

459 and loadopt.local_opts.get(

460 "undefer_group_%s" % self.group, False

461 )

462 )

463 or (only_load_props and self.key in only_load_props)

464 ):

465 self.parent_property._get_strategy(

466 (("deferred", False), ("instrument", True))

467 ).setup_query(

468 compile_state,

469 query_entity,

470 path,

471 loadopt,

472 adapter,

473 column_collection,

474 memoized_populators,

475 **kw

476 )

477 elif self.is_class_level:

478 memoized_populators[self.parent_property] = _SET_DEFERRED_EXPIRED

479 elif not self.raiseload:

480 memoized_populators[self.parent_property] = _DEFER_FOR_STATE

481 else:

482 memoized_populators[self.parent_property] = _RAISE_FOR_STATE

483

484 def _load_for_state(self, state, passive):

485 if not state.key:

486 return attributes.ATTR_EMPTY

487

488 if not passive & attributes.SQL_OK:

489 return attributes.PASSIVE_NO_RESULT

490

491 localparent = state.manager.mapper

492

493 if self.group:

494 toload = [

495 p.key

496 for p in localparent.iterate_properties

497 if isinstance(p, StrategizedProperty)

498 and isinstance(p.strategy, DeferredColumnLoader)

499 and p.group == self.group

500 ]

501 else:

502 toload = [self.key]

503

504 # narrow the keys down to just those which have no history

505 group = [k for k in toload if k in state.unmodified]

506

507 session = _state_session(state)

508 if session is None:

509 raise orm_exc.DetachedInstanceError(

510 "Parent instance %s is not bound to a Session; "

511 "deferred load operation of attribute '%s' cannot proceed"

512 % (orm_util.state_str(state), self.key)

513 )

514

515 if self.raiseload:

516 self._invoke_raise_load(state, passive, "raise")

517

518 if (

519 loading.load_on_ident(

520 session,

521 sql.select(localparent).set_label_style(

522 LABEL_STYLE_TABLENAME_PLUS_COL

523 ),

524 state.key,

525 only_load_props=group,

526 refresh_state=state,

527 )

528 is None

529 ):

530 raise orm_exc.ObjectDeletedError(state)

531

532 return attributes.ATTR_WAS_SET

533

534 def _invoke_raise_load(self, state, passive, lazy):

535 raise sa_exc.InvalidRequestError(

536 "'%s' is not available due to raiseload=True" % (self,)

537 )

538

539

540class LoadDeferredColumns(object):

541 """serializable loader object used by DeferredColumnLoader"""

542

543 def __init__(self, key, raiseload=False):

544 self.key = key

545 self.raiseload = raiseload

546

547 def __call__(self, state, passive=attributes.PASSIVE_OFF):

548 key = self.key

549

550 localparent = state.manager.mapper

551 prop = localparent._props[key]

552 if self.raiseload:

553 strategy_key = (

554 ("deferred", True),

555 ("instrument", True),

556 ("raiseload", True),

557 )

558 else:

559 strategy_key = (("deferred", True), ("instrument", True))

560 strategy = prop._get_strategy(strategy_key)

561 return strategy._load_for_state(state, passive)

562

563

564class AbstractRelationshipLoader(LoaderStrategy):

565 """LoaderStratgies which deal with related objects."""

566

567 __slots__ = "mapper", "target", "uselist", "entity"

568

569 def __init__(self, parent, strategy_key):

570 super(AbstractRelationshipLoader, self).__init__(parent, strategy_key)

571 self.mapper = self.parent_property.mapper

572 self.entity = self.parent_property.entity

573 self.target = self.parent_property.target

574 self.uselist = self.parent_property.uselist

575

576

577@log.class_logger

578@relationships.RelationshipProperty.strategy_for(do_nothing=True)

579class DoNothingLoader(LoaderStrategy):

580 """Relationship loader that makes no change to the object's state.

581

582 Compared to NoLoader, this loader does not initialize the

583 collection/attribute to empty/none; the usual default LazyLoader will

584 take effect.

585

586 """

587

588

589@log.class_logger

590@relationships.RelationshipProperty.strategy_for(lazy="noload")

591@relationships.RelationshipProperty.strategy_for(lazy=None)

592class NoLoader(AbstractRelationshipLoader):

593 """Provide loading behavior for a :class:`.RelationshipProperty`

594 with "lazy=None".

595

596 """

597

598 __slots__ = ()

599

600 def init_class_attribute(self, mapper):

601 self.is_class_level = True

602

603 _register_attribute(

604 self.parent_property,

605 mapper,

606 useobject=True,

607 typecallable=self.parent_property.collection_class,

608 )

609

610 def create_row_processor(

611 self,

612 context,

613 query_entity,

614 path,

615 loadopt,

616 mapper,

617 result,

618 adapter,

619 populators,

620 ):

621 def invoke_no_load(state, dict_, row):

622 if self.uselist:

623 attributes.init_state_collection(state, dict_, self.key)

624 else:

625 dict_[self.key] = None

626

627 populators["new"].append((self.key, invoke_no_load))

628

629

630@log.class_logger

631@relationships.RelationshipProperty.strategy_for(lazy=True)

632@relationships.RelationshipProperty.strategy_for(lazy="select")

633@relationships.RelationshipProperty.strategy_for(lazy="raise")

634@relationships.RelationshipProperty.strategy_for(lazy="raise_on_sql")

635@relationships.RelationshipProperty.strategy_for(lazy="baked_select")

636class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):

637 """Provide loading behavior for a :class:`.RelationshipProperty`

638 with "lazy=True", that is loads when first accessed.

639

640 """

641

642 __slots__ = (

643 "_lazywhere",

644 "_rev_lazywhere",

645 "_lazyload_reverse_option",

646 "_order_by",

647 "use_get",

648 "is_aliased_class",

649 "_bind_to_col",

650 "_equated_columns",

651 "_rev_bind_to_col",

652 "_rev_equated_columns",

653 "_simple_lazy_clause",

654 "_raise_always",

655 "_raise_on_sql",

656 )

657

658 def __init__(self, parent, strategy_key):

659 super(LazyLoader, self).__init__(parent, strategy_key)

660 self._raise_always = self.strategy_opts["lazy"] == "raise"

661 self._raise_on_sql = self.strategy_opts["lazy"] == "raise_on_sql"

662

663 self.is_aliased_class = inspect(self.entity).is_aliased_class

664

665 join_condition = self.parent_property._join_condition

666 (

667 self._lazywhere,

668 self._bind_to_col,

669 self._equated_columns,

670 ) = join_condition.create_lazy_clause()

671

672 (

673 self._rev_lazywhere,

674 self._rev_bind_to_col,

675 self._rev_equated_columns,

676 ) = join_condition.create_lazy_clause(reverse_direction=True)

677

678 if self.parent_property.order_by:

679 self._order_by = [

680 sql_util._deep_annotate(elem, {"_orm_adapt": True})

681 for elem in util.to_list(self.parent_property.order_by)

682 ]

683 else:

684 self._order_by = None

685

686 self.logger.info("%s lazy loading clause %s", self, self._lazywhere)

687

688 # determine if our "lazywhere" clause is the same as the mapper's

689 # get() clause. then we can just use mapper.get()

690 #

691 # TODO: the "not self.uselist" can be taken out entirely; a m2o

692 # load that populates for a list (very unusual, but is possible with

693 # the API) can still set for "None" and the attribute system will

694 # populate as an empty list.

695 self.use_get = (

696 not self.is_aliased_class

697 and not self.uselist

698 and self.entity._get_clause[0].compare(

699 self._lazywhere,

700 use_proxies=True,

701 compare_keys=False,

702 equivalents=self.mapper._equivalent_columns,

703 )

704 )

705

706 if self.use_get:

707 for col in list(self._equated_columns):

708 if col in self.mapper._equivalent_columns:

709 for c in self.mapper._equivalent_columns[col]:

710 self._equated_columns[c] = self._equated_columns[col]

711

712 self.logger.info(

713 "%s will use Session.get() to " "optimize instance loads", self

714 )

715

716 def init_class_attribute(self, mapper):

717 self.is_class_level = True

718

719 _legacy_inactive_history_style = (

720 self.parent_property._legacy_inactive_history_style

721 )

722

723 if self.parent_property.active_history:

724 active_history = True

725 _deferred_history = False

726

727 elif (

728 self.parent_property.direction is not interfaces.MANYTOONE

729 or not self.use_get

730 ):

731 if _legacy_inactive_history_style:

732 active_history = True

733 _deferred_history = False

734 else:

735 active_history = False

736 _deferred_history = True

737 else:

738 active_history = _deferred_history = False

739

740 _register_attribute(

741 self.parent_property,

742 mapper,

743 useobject=True,

744 callable_=self._load_for_state,

745 typecallable=self.parent_property.collection_class,

746 active_history=active_history,

747 _deferred_history=_deferred_history,

748 )

749

750 def _memoized_attr__simple_lazy_clause(self):

751

752 lazywhere = sql_util._deep_annotate(

753 self._lazywhere, {"_orm_adapt": True}

754 )

755

756 criterion, bind_to_col = (lazywhere, self._bind_to_col)

757

758 params = []

759

760 def visit_bindparam(bindparam):

761 bindparam.unique = False

762

763 visitors.traverse(criterion, {}, {"bindparam": visit_bindparam})

764

765 def visit_bindparam(bindparam):

766 if bindparam._identifying_key in bind_to_col:

767 params.append(

768 (

769 bindparam.key,

770 bind_to_col[bindparam._identifying_key],

771 None,

772 )

773 )

774 elif bindparam.callable is None:

775 params.append((bindparam.key, None, bindparam.value))

776

777 criterion = visitors.cloned_traverse(

778 criterion, {}, {"bindparam": visit_bindparam}

779 )

780

781 return criterion, params

782

783 def _generate_lazy_clause(self, state, passive):

784 criterion, param_keys = self._simple_lazy_clause

785

786 if state is None:

787 return sql_util.adapt_criterion_to_null(

788 criterion, [key for key, ident, value in param_keys]

789 )

790

791 mapper = self.parent_property.parent

792

793 o = state.obj() # strong ref

794 dict_ = attributes.instance_dict(o)

795

796 if passive & attributes.INIT_OK:

797 passive ^= attributes.INIT_OK

798

799 params = {}

800 for key, ident, value in param_keys:

801 if ident is not None:

802 if passive and passive & attributes.LOAD_AGAINST_COMMITTED:

803 value = mapper._get_committed_state_attr_by_column(

804 state, dict_, ident, passive

805 )

806 else:

807 value = mapper._get_state_attr_by_column(

808 state, dict_, ident, passive

809 )

810

811 params[key] = value

812

813 return criterion, params

814

815 def _invoke_raise_load(self, state, passive, lazy):

816 raise sa_exc.InvalidRequestError(

817 "'%s' is not available due to lazy='%s'" % (self, lazy)

818 )

819

820 def _load_for_state(self, state, passive, loadopt=None, extra_criteria=()):

821 if not state.key and (

822 (

823 not self.parent_property.load_on_pending

824 and not state._load_pending

825 )

826 or not state.session_id

827 ):

828 return attributes.ATTR_EMPTY

829

830 pending = not state.key

831 primary_key_identity = None

832

833 use_get = self.use_get and (not loadopt or not loadopt._extra_criteria)

834

835 if (not passive & attributes.SQL_OK and not use_get) or (

836 not passive & attributes.NON_PERSISTENT_OK and pending

837 ):

838 return attributes.PASSIVE_NO_RESULT

839

840 if (

841 # we were given lazy="raise"

842 self._raise_always

843 # the no_raise history-related flag was not passed

844 and not passive & attributes.NO_RAISE

845 and (

846 # if we are use_get and related_object_ok is disabled,

847 # which means we are at most looking in the identity map

848 # for history purposes or otherwise returning

849 # PASSIVE_NO_RESULT, don't raise. This is also a

850 # history-related flag

851 not use_get

852 or passive & attributes.RELATED_OBJECT_OK

853 )

854 ):

855

856 self._invoke_raise_load(state, passive, "raise")

857

858 session = _state_session(state)

859 if not session:

860 if passive & attributes.NO_RAISE:

861 return attributes.PASSIVE_NO_RESULT

862

863 raise orm_exc.DetachedInstanceError(

864 "Parent instance %s is not bound to a Session; "

865 "lazy load operation of attribute '%s' cannot proceed"

866 % (orm_util.state_str(state), self.key)

867 )

868

869 # if we have a simple primary key load, check the

870 # identity map without generating a Query at all

871 if use_get:

872 primary_key_identity = self._get_ident_for_use_get(

873 session, state, passive

874 )

875 if attributes.PASSIVE_NO_RESULT in primary_key_identity:

876 return attributes.PASSIVE_NO_RESULT

877 elif attributes.NEVER_SET in primary_key_identity:

878 return attributes.NEVER_SET

879

880 if _none_set.issuperset(primary_key_identity):

881 return None

882

883 if (

884 self.key in state.dict

885 and not passive & attributes.DEFERRED_HISTORY_LOAD

886 ):

887 return attributes.ATTR_WAS_SET

888

889 # look for this identity in the identity map. Delegate to the

890 # Query class in use, as it may have special rules for how it

891 # does this, including how it decides what the correct

892 # identity_token would be for this identity.

893

894 instance = session._identity_lookup(

895 self.entity,

896 primary_key_identity,

897 passive=passive,

898 lazy_loaded_from=state,

899 )

900

901 if instance is not None:

902 if instance is attributes.PASSIVE_CLASS_MISMATCH:

903 return None

904 else:

905 return instance

906 elif (

907 not passive & attributes.SQL_OK

908 or not passive & attributes.RELATED_OBJECT_OK

909 ):

910 return attributes.PASSIVE_NO_RESULT

911

912 return self._emit_lazyload(

913 session,

914 state,

915 primary_key_identity,

916 passive,

917 loadopt,

918 extra_criteria,

919 )

920

921 def _get_ident_for_use_get(self, session, state, passive):

922 instance_mapper = state.manager.mapper

923

924 if passive & attributes.LOAD_AGAINST_COMMITTED:

925 get_attr = instance_mapper._get_committed_state_attr_by_column

926 else:

927 get_attr = instance_mapper._get_state_attr_by_column

928

929 dict_ = state.dict

930

931 return [

932 get_attr(state, dict_, self._equated_columns[pk], passive=passive)

933 for pk in self.mapper.primary_key

934 ]

935

936 @util.preload_module("sqlalchemy.orm.strategy_options")

937 def _emit_lazyload(

938 self,

939 session,

940 state,

941 primary_key_identity,

942 passive,

943 loadopt,

944 extra_criteria,

945 ):

946 strategy_options = util.preloaded.orm_strategy_options

947

948 clauseelement = self.entity.__clause_element__()

949 stmt = Select._create_raw_select(

950 _raw_columns=[clauseelement],

951 _propagate_attrs=clauseelement._propagate_attrs,

952 _label_style=LABEL_STYLE_TABLENAME_PLUS_COL,

953 _compile_options=ORMCompileState.default_compile_options,

954 )

955 load_options = QueryContext.default_load_options

956

957 load_options += {

958 "_invoke_all_eagers": False,

959 "_lazy_loaded_from": state,

960 }

961

962 if self.parent_property.secondary is not None:

963 stmt = stmt.select_from(

964 self.mapper, self.parent_property.secondary

965 )

966

967 pending = not state.key

968

969 # don't autoflush on pending

970 if pending or passive & attributes.NO_AUTOFLUSH:

971 stmt._execution_options = util.immutabledict({"autoflush": False})

972

973 use_get = self.use_get

974

975 if state.load_options or (loadopt and loadopt._extra_criteria):

976 effective_path = state.load_path[self.parent_property]

977

978 opts = tuple(state.load_options)

979

980 if loadopt and loadopt._extra_criteria:

981 use_get = False

982 opts += (

983 orm_util.LoaderCriteriaOption(self.entity, extra_criteria),

984 )

985 stmt._with_options = opts

986 else:

987 # this path is used if there are not already any options

988 # in the query, but an event may want to add them

989 effective_path = state.mapper._path_registry[self.parent_property]

990 stmt._compile_options += {"_current_path": effective_path}

991

992 if use_get:

993 if self._raise_on_sql and not passive & attributes.NO_RAISE:

994 self._invoke_raise_load(state, passive, "raise_on_sql")

995

996 return loading.load_on_pk_identity(

997 session, stmt, primary_key_identity, load_options=load_options

998 )

999

1000 if self._order_by:

1001 stmt._order_by_clauses = self._order_by

1002

1003 def _lazyload_reverse(compile_context):

1004 for rev in self.parent_property._reverse_property:

1005 # reverse props that are MANYTOONE are loading *this*

1006 # object from get(), so don't need to eager out to those.

1007 if (

1008 rev.direction is interfaces.MANYTOONE

1009 and rev._use_get

1010 and not isinstance(rev.strategy, LazyLoader)

1011 ):

1012 strategy_options.Load.for_existing_path(

1013 compile_context.compile_options._current_path[

1014 rev.parent

1015 ]

1016 ).lazyload(rev).process_compile_state(compile_context)

1017

1018 stmt._with_context_options += (

1019 (_lazyload_reverse, self.parent_property),

1020 )

1021

1022 lazy_clause, params = self._generate_lazy_clause(state, passive)

1023

1024 execution_options = {

1025 "_sa_orm_load_options": load_options,

1026 }

1027

1028 if (

1029 self.key in state.dict

1030 and not passive & attributes.DEFERRED_HISTORY_LOAD

1031 ):

1032 return attributes.ATTR_WAS_SET

1033

1034 if pending:

1035 if util.has_intersection(orm_util._none_set, params.values()):

1036 return None

1037

1038 elif util.has_intersection(orm_util._never_set, params.values()):

1039 return None

1040

1041 if self._raise_on_sql and not passive & attributes.NO_RAISE:

1042 self._invoke_raise_load(state, passive, "raise_on_sql")

1043

1044 stmt._where_criteria = (lazy_clause,)

1045

1046 result = session.execute(

1047 stmt, params, execution_options=execution_options

1048 )

1049

1050 result = result.unique().scalars().all()

1051

1052 if self.uselist:

1053 return result

1054 else:

1055 l = len(result)

1056 if l:

1057 if l > 1:

1058 util.warn(

1059 "Multiple rows returned with "

1060 "uselist=False for lazily-loaded attribute '%s' "

1061 % self.parent_property

1062 )

1063

1064 return result[0]

1065 else:

1066 return None

1067

1068 def create_row_processor(

1069 self,

1070 context,

1071 query_entity,

1072 path,

1073 loadopt,

1074 mapper,

1075 result,

1076 adapter,

1077 populators,

1078 ):

1079 key = self.key

1080

1081 if not self.is_class_level or (loadopt and loadopt._extra_criteria):

1082 # we are not the primary manager for this attribute

1083 # on this class - set up a

1084 # per-instance lazyloader, which will override the

1085 # class-level behavior.

1086 # this currently only happens when using a

1087 # "lazyload" option on a "no load"

1088 # attribute - "eager" attributes always have a

1089 # class-level lazyloader installed.

1090 set_lazy_callable = (

1091 InstanceState._instance_level_callable_processor

1092 )(

1093 mapper.class_manager,

1094 LoadLazyAttribute(

1095 key,

1096 self,

1097 loadopt,

1098 loadopt._generate_extra_criteria(context)

1099 if loadopt._extra_criteria

1100 else None,

1101 ),

1102 key,

1103 )

1104

1105 populators["new"].append((self.key, set_lazy_callable))

1106 elif context.populate_existing or mapper.always_refresh:

1107

1108 def reset_for_lazy_callable(state, dict_, row):

1109 # we are the primary manager for this attribute on

1110 # this class - reset its

1111 # per-instance attribute state, so that the class-level

1112 # lazy loader is

1113 # executed when next referenced on this instance.

1114 # this is needed in

1115 # populate_existing() types of scenarios to reset

1116 # any existing state.

1117 state._reset(dict_, key)

1118

1119 populators["new"].append((self.key, reset_for_lazy_callable))

1120

1121

1122class LoadLazyAttribute(object):

1123 """semi-serializable loader object used by LazyLoader

1124

1125 Historically, this object would be carried along with instances that

1126 needed to run lazyloaders, so it had to be serializable to support

1127 cached instances.

1128

1129 this is no longer a general requirement, and the case where this object

1130 is used is exactly the case where we can't really serialize easily,

1131 which is when extra criteria in the loader option is present.

1132

1133 We can't reliably serialize that as it refers to mapped entities and

1134 AliasedClass objects that are local to the current process, which would

1135 need to be matched up on deserialize e.g. the sqlalchemy.ext.serializer

1136 approach.

1137

1138 """

1139

1140 def __init__(self, key, initiating_strategy, loadopt, extra_criteria):

1141 self.key = key

1142 self.strategy_key = initiating_strategy.strategy_key

1143 self.loadopt = loadopt

1144 self.extra_criteria = extra_criteria

1145

1146 def __getstate__(self):

1147 if self.extra_criteria is not None:

1148 util.warn(

1149 "Can't reliably serialize a lazyload() option that "

1150 "contains additional criteria; please use eager loading "

1151 "for this case"

1152 )

1153 return {

1154 "key": self.key,

1155 "strategy_key": self.strategy_key,

1156 "loadopt": self.loadopt,

1157 "extra_criteria": (),

1158 }

1159

1160 def __call__(self, state, passive=attributes.PASSIVE_OFF):

1161 key = self.key

1162 instance_mapper = state.manager.mapper

1163 prop = instance_mapper._props[key]

1164 strategy = prop._strategies[self.strategy_key]

1165

1166 return strategy._load_for_state(

1167 state,

1168 passive,

1169 loadopt=self.loadopt,

1170 extra_criteria=self.extra_criteria,

1171 )

1172

1173

1174class PostLoader(AbstractRelationshipLoader):

1175 """A relationship loader that emits a second SELECT statement."""

1176

1177 def _check_recursive_postload(self, context, path, join_depth=None):

1178 effective_path = (

1179 context.compile_state.current_path or orm_util.PathRegistry.root

1180 ) + path

1181

1182 if loading.PostLoad.path_exists(

1183 context, effective_path, self.parent_property

1184 ):

1185 return True

1186

1187 path_w_prop = path[self.parent_property]

1188 effective_path_w_prop = effective_path[self.parent_property]

1189

1190 if not path_w_prop.contains(context.attributes, "loader"):

1191 if join_depth:

1192 if effective_path_w_prop.length / 2 > join_depth:

1193 return True

1194 elif effective_path_w_prop.contains_mapper(self.mapper):

1195 return True

1196

1197 return False

1198

1199 def _immediateload_create_row_processor(

1200 self,

1201 context,

1202 query_entity,

1203 path,

1204 loadopt,

1205 mapper,

1206 result,

1207 adapter,

1208 populators,

1209 ):

1210 return self.parent_property._get_strategy(

1211 (("lazy", "immediate"),)

1212 ).create_row_processor(

1213 context,

1214 query_entity,

1215 path,

1216 loadopt,

1217 mapper,

1218 result,

1219 adapter,

1220 populators,

1221 )

1222

1223

1224@relationships.RelationshipProperty.strategy_for(lazy="immediate")

1225class ImmediateLoader(PostLoader):

1226 __slots__ = ()

1227

1228 def init_class_attribute(self, mapper):

1229 self.parent_property._get_strategy(

1230 (("lazy", "select"),)

1231 ).init_class_attribute(mapper)

1232

1233 def create_row_processor(

1234 self,

1235 context,

1236 query_entity,

1237 path,

1238 loadopt,

1239 mapper,

1240 result,

1241 adapter,

1242 populators,

1243 ):

1244 def load_immediate(state, dict_, row):

1245 state.get_impl(self.key).get(state, dict_, flags)

1246

1247 if self._check_recursive_postload(context, path):

1248 # this will not emit SQL and will only emit for a many-to-one

1249 # "use get" load. the "_RELATED" part means it may return

1250 # instance even if its expired, since this is a mutually-recursive

1251 # load operation.

1252 flags = attributes.PASSIVE_NO_FETCH_RELATED | attributes.NO_RAISE

1253 else:

1254 flags = attributes.PASSIVE_OFF | attributes.NO_RAISE

1255

1256 populators["delayed"].append((self.key, load_immediate))

1257

1258

1259@log.class_logger

1260@relationships.RelationshipProperty.strategy_for(lazy="subquery")

1261class SubqueryLoader(PostLoader):

1262 __slots__ = ("join_depth",)

1263

1264 def __init__(self, parent, strategy_key):

1265 super(SubqueryLoader, self).__init__(parent, strategy_key)

1266 self.join_depth = self.parent_property.join_depth

1267

1268 def init_class_attribute(self, mapper):

1269 self.parent_property._get_strategy(

1270 (("lazy", "select"),)

1271 ).init_class_attribute(mapper)

1272

1273 def _get_leftmost(

1274 self,

1275 orig_query_entity_index,

1276 subq_path,

1277 current_compile_state,

1278 is_root,

1279 ):

1280 given_subq_path = subq_path

1281 subq_path = subq_path.path

1282 subq_mapper = orm_util._class_to_mapper(subq_path[0])

1283

1284 # determine attributes of the leftmost mapper

1285 if (

1286 self.parent.isa(subq_mapper)

1287 and self.parent_property is subq_path[1]

1288 ):

1289 leftmost_mapper, leftmost_prop = self.parent, self.parent_property

1290 else:

1291 leftmost_mapper, leftmost_prop = subq_mapper, subq_path[1]

1292

1293 if is_root:

1294 # the subq_path is also coming from cached state, so when we start

1295 # building up this path, it has to also be converted to be in terms

1296 # of the current state. this is for the specific case of the entity

1297 # is an AliasedClass against a subquery that's not otherwise going

1298 # to adapt

1299 new_subq_path = current_compile_state._entities[

1300 orig_query_entity_index

1301 ].entity_zero._path_registry[leftmost_prop]

1302 additional = len(subq_path) - len(new_subq_path)

1303 if additional:

1304 new_subq_path += path_registry.PathRegistry.coerce(

1305 subq_path[-additional:]

1306 )

1307 else:

1308 new_subq_path = given_subq_path

1309

1310 leftmost_cols = leftmost_prop.local_columns

1311

1312 leftmost_attr = [

1313 getattr(

1314 new_subq_path.path[0].entity,

1315 leftmost_mapper._columntoproperty[c].key,

1316 )

1317 for c in leftmost_cols

1318 ]

1319

1320 return leftmost_mapper, leftmost_attr, leftmost_prop, new_subq_path

1321

1322 def _generate_from_original_query(

1323 self,

1324 orig_compile_state,

1325 orig_query,

1326 leftmost_mapper,

1327 leftmost_attr,

1328 leftmost_relationship,

1329 orig_entity,

1330 ):

1331 # reformat the original query

1332 # to look only for significant columns

1333 q = orig_query._clone().correlate(None)

1334

1335 # LEGACY: make a Query back from the select() !!

1336 # This suits at least two legacy cases:

1337 # 1. applications which expect before_compile() to be called

1338 # below when we run .subquery() on this query (Keystone)

1339 # 2. applications which are doing subqueryload with complex

1340 # from_self() queries, as query.subquery() / .statement

1341 # has to do the full compile context for multiply-nested

1342 # from_self() (Neutron) - see test_subqload_from_self

1343 # for demo.

1344 q2 = query.Query.__new__(query.Query)

1345 q2.__dict__.update(q.__dict__)

1346 q = q2

1347

1348 # set the query's "FROM" list explicitly to what the

1349 # FROM list would be in any case, as we will be limiting

1350 # the columns in the SELECT list which may no longer include

1351 # all entities mentioned in things like WHERE, JOIN, etc.

1352 if not q._from_obj:

1353 q._enable_assertions = False

1354 q.select_from.non_generative(

1355 q,

1356 *{

1357 ent["entity"]

1358 for ent in _column_descriptions(

1359 orig_query, compile_state=orig_compile_state

1360 )

1361 if ent["entity"] is not None

1362 }

1363 )

1364

1365 # select from the identity columns of the outer (specifically, these

1366 # are the 'local_cols' of the property). This will remove other

1367 # columns from the query that might suggest the right entity which is

1368 # why we do set select_from above. The attributes we have are

1369 # coerced and adapted using the original query's adapter, which is

1370 # needed only for the case of adapting a subclass column to

1371 # that of a polymorphic selectable, e.g. we have

1372 # Engineer.primary_language and the entity is Person. All other

1373 # adaptations, e.g. from_self, select_entity_from(), will occur

1374 # within the new query when it compiles, as the compile_state we are

1375 # using here is only a partial one. If the subqueryload is from a

1376 # with_polymorphic() or other aliased() object, left_attr will already

1377 # be the correct attributes so no adaptation is needed.

1378 target_cols = orig_compile_state._adapt_col_list(

1379 [

1380 sql.coercions.expect(sql.roles.ColumnsClauseRole, o)

1381 for o in leftmost_attr

1382 ],

1383 orig_compile_state._get_current_adapter(),

1384 )

1385 q._raw_columns = target_cols

1386

1387 distinct_target_key = leftmost_relationship.distinct_target_key

1388

1389 if distinct_target_key is True:

1390 q._distinct = True

1391 elif distinct_target_key is None:

1392 # if target_cols refer to a non-primary key or only

1393 # part of a composite primary key, set the q as distinct

1394 for t in set(c.table for c in target_cols):

1395 if not set(target_cols).issuperset(t.primary_key):

1396 q._distinct = True

1397 break

1398

1399 # don't need ORDER BY if no limit/offset

1400 if not q._has_row_limiting_clause:

1401 q._order_by_clauses = ()

1402

1403 if q._distinct is True and q._order_by_clauses:

1404 # the logic to automatically add the order by columns to the query

1405 # when distinct is True is deprecated in the query

1406 to_add = sql_util.expand_column_list_from_order_by(

1407 target_cols, q._order_by_clauses

1408 )

1409 if to_add:

1410 q._set_entities(target_cols + to_add)

1411

1412 # the original query now becomes a subquery

1413 # which we'll join onto.

1414 # LEGACY: as "q" is a Query, the before_compile() event is invoked

1415 # here.

1416 embed_q = q.set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL).subquery()

1417 left_alias = orm_util.AliasedClass(

1418 leftmost_mapper, embed_q, use_mapper_path=True

1419 )

1420 return left_alias

1421

1422 def _prep_for_joins(self, left_alias, subq_path):

1423 # figure out what's being joined. a.k.a. the fun part

1424 to_join = []

1425 pairs = list(subq_path.pairs())

1426

1427 for i, (mapper, prop) in enumerate(pairs):

1428 if i > 0:

1429 # look at the previous mapper in the chain -

1430 # if it is as or more specific than this prop's

1431 # mapper, use that instead.

1432 # note we have an assumption here that

1433 # the non-first element is always going to be a mapper,

1434 # not an AliasedClass

1435

1436 prev_mapper = pairs[i - 1][1].mapper

1437 to_append = prev_mapper if prev_mapper.isa(mapper) else mapper

1438 else:

1439 to_append = mapper

1440

1441 to_join.append((to_append, prop.key))

1442

1443 # determine the immediate parent class we are joining from,

1444 # which needs to be aliased.

1445

1446 if len(to_join) < 2:

1447 # in the case of a one level eager load, this is the

1448 # leftmost "left_alias".

1449 parent_alias = left_alias

1450 else:

1451 info = inspect(to_join[-1][0])

1452 if info.is_aliased_class:

1453 parent_alias = info.entity

1454 else:

1455 # alias a plain mapper as we may be

1456 # joining multiple times

1457 parent_alias = orm_util.AliasedClass(

1458 info.entity, use_mapper_path=True

1459 )

1460

1461 local_cols = self.parent_property.local_columns

1462

1463 local_attr = [

1464 getattr(parent_alias, self.parent._columntoproperty[c].key)

1465 for c in local_cols

1466 ]

1467 return to_join, local_attr, parent_alias

1468

1469 def _apply_joins(

1470 self, q, to_join, left_alias, parent_alias, effective_entity

1471 ):

1472

1473 ltj = len(to_join)

1474 if ltj == 1:

1475 to_join = [

1476 getattr(left_alias, to_join[0][1]).of_type(effective_entity)

1477 ]

1478 elif ltj == 2:

1479 to_join = [

1480 getattr(left_alias, to_join[0][1]).of_type(parent_alias),

1481 getattr(parent_alias, to_join[-1][1]).of_type(

1482 effective_entity

1483 ),

1484 ]

1485 elif ltj > 2:

1486 middle = [

1487 (

1488 orm_util.AliasedClass(item[0])

1489 if not inspect(item[0]).is_aliased_class

1490 else item[0].entity,

1491 item[1],

1492 )

1493 for item in to_join[1:-1]

1494 ]

1495 inner = []

1496

1497 while middle:

1498 item = middle.pop(0)

1499 attr = getattr(item[0], item[1])

1500 if middle:

1501 attr = attr.of_type(middle[0][0])

1502 else:

1503 attr = attr.of_type(parent_alias)

1504

1505 inner.append(attr)

1506

1507 to_join = (

1508 [getattr(left_alias, to_join[0][1]).of_type(inner[0].parent)]

1509 + inner

1510 + [

1511 getattr(parent_alias, to_join[-1][1]).of_type(

1512 effective_entity

1513 )

1514 ]

1515 )

1516

1517 for attr in to_join:

1518 q = q.join(attr)

1519

1520 return q

1521

1522 def _setup_options(

1523 self,

1524 context,

1525 q,

1526 subq_path,

1527 rewritten_path,

1528 orig_query,

1529 effective_entity,

1530 loadopt,

1531 ):

1532

1533 # note that because the subqueryload object

1534 # does not re-use the cached query, instead always making

1535 # use of the current invoked query, while we have two queries

1536 # here (orig and context.query), they are both non-cached

1537 # queries and we can transfer the options as is without

1538 # adjusting for new criteria. Some work on #6881 / #6889

1539 # brought this into question.

1540 new_options = orig_query._with_options

1541

1542 if loadopt and loadopt._extra_criteria:

1543

1544 new_options += (

1545 orm_util.LoaderCriteriaOption(

1546 self.entity,

1547 loadopt._generate_extra_criteria(context),

1548 ),

1549 )

1550

1551 # propagate loader options etc. to the new query.

1552 # these will fire relative to subq_path.

1553 q = q._with_current_path(rewritten_path)

1554 q = q.options(*new_options)

1555

1556 return q

1557

1558 def _setup_outermost_orderby(self, q):

1559 if self.parent_property.order_by:

1560

1561 def _setup_outermost_orderby(compile_context):

1562 compile_context.eager_order_by += tuple(

1563 util.to_list(self.parent_property.order_by)

1564 )

1565

1566 q = q._add_context_option(

1567 _setup_outermost_orderby, self.parent_property

1568 )

1569

1570 return q

1571

1572 class _SubqCollections(object):

1573 """Given a :class:`_query.Query` used to emit the "subquery load",

1574 provide a load interface that executes the query at the

1575 first moment a value is needed.

1576

1577 """

1578

1579 __slots__ = (

1580 "session",

1581 "execution_options",

1582 "load_options",

1583 "params",

1584 "subq",

1585 "_data",

1586 )

1587

1588 def __init__(self, context, subq):

1589 # avoid creating a cycle by storing context

1590 # even though that's preferable

1591 self.session = context.session

1592 self.execution_options = context.execution_options

1593 self.load_options = context.load_options

1594 self.params = context.params or {}

1595 self.subq = subq

1596 self._data = None

1597

1598 def get(self, key, default):

1599 if self._data is None:

1600 self._load()

1601 return self._data.get(key, default)

1602

1603 def _load(self):

1604 self._data = collections.defaultdict(list)

1605

1606 q = self.subq

1607 assert q.session is None

1608

1609 q = q.with_session(self.session)

1610

1611 if self.load_options._populate_existing:

1612 q = q.populate_existing()

1613 # to work with baked query, the parameters may have been

1614 # updated since this query was created, so take these into account

1615

1616 rows = list(q.params(self.params))

1617 for k, v in itertools.groupby(rows, lambda x: x[1:]):

1618 self._data[k].extend(vv[0] for vv in v)

1619

1620 def loader(self, state, dict_, row):

1621 if self._data is None:

1622 self._load()

1623

1624 def _setup_query_from_rowproc(

1625 self,

1626 context,

1627 query_entity,

1628 path,

1629 entity,

1630 loadopt,

1631 adapter,

1632 ):

1633 compile_state = context.compile_state

1634 if (

1635 not compile_state.compile_options._enable_eagerloads

1636 or compile_state.compile_options._for_refresh_state

1637 ):

1638 return

1639

1640 orig_query_entity_index = compile_state._entities.index(query_entity)

1641 context.loaders_require_buffering = True

1642

1643 path = path[self.parent_property]

1644

1645 # build up a path indicating the path from the leftmost

1646 # entity to the thing we're subquery loading.

1647 with_poly_entity = path.get(

1648 compile_state.attributes, "path_with_polymorphic", None

1649 )

1650 if with_poly_entity is not None:

1651 effective_entity = with_poly_entity

1652 else:

1653 effective_entity = self.entity

1654

1655 subq_path, rewritten_path = context.query._execution_options.get(

1656 ("subquery_paths", None),

1657 (orm_util.PathRegistry.root, orm_util.PathRegistry.root),

1658 )

1659 is_root = subq_path is orm_util.PathRegistry.root

1660 subq_path = subq_path + path

1661 rewritten_path = rewritten_path + path

1662

1663 # if not via query option, check for

1664 # a cycle

1665 # TODO: why is this here??? this is now handled

1666 # by the _check_recursive_postload call

1667 if not path.contains(compile_state.attributes, "loader"):

1668 if self.join_depth:

1669 if (

1670 (

1671 compile_state.current_path.length

1672 if compile_state.current_path

1673 else 0

1674 )

1675 + path.length

1676 ) / 2 > self.join_depth:

1677 return

1678 elif subq_path.contains_mapper(self.mapper):

1679 return

1680

1681 # use the current query being invoked, not the compile state

1682 # one. this is so that we get the current parameters. however,

1683 # it means we can't use the existing compile state, we have to make

1684 # a new one. other approaches include possibly using the

1685 # compiled query but swapping the params, seems only marginally

1686 # less time spent but more complicated

1687 orig_query = context.query._execution_options.get(

1688 ("orig_query", SubqueryLoader), context.query

1689 )

1690

1691 # make a new compile_state for the query that's probably cached, but

1692 # we're sort of undoing a bit of that caching :(

1693 compile_state_cls = ORMCompileState._get_plugin_class_for_plugin(

1694 orig_query, "orm"

1695 )

1696

1697 if orig_query._is_lambda_element:

1698 if context.load_options._lazy_loaded_from is None:

1699 util.warn(

1700 'subqueryloader for "%s" must invoke lambda callable '

1701 "at %r in "

1702 "order to produce a new query, decreasing the efficiency "

1703 "of caching for this statement. Consider using "

1704 "selectinload() for more effective full-lambda caching"

1705 % (self, orig_query)

1706 )

1707 orig_query = orig_query._resolved

1708

1709 # this is the more "quick" version, however it's not clear how

1710 # much of this we need. in particular I can't get a test to

1711 # fail if the "set_base_alias" is missing and not sure why that is.

1712 orig_compile_state = compile_state_cls._create_entities_collection(

1713 orig_query, legacy=False

1714 )

1715

1716 (

1717 leftmost_mapper,

1718 leftmost_attr,

1719 leftmost_relationship,

1720 rewritten_path,

1721 ) = self._get_leftmost(

1722 orig_query_entity_index,

1723 rewritten_path,

1724 orig_compile_state,

1725 is_root,

1726 )

1727

1728 # generate a new Query from the original, then

1729 # produce a subquery from it.

1730 left_alias = self._generate_from_original_query(

1731 orig_compile_state,

1732 orig_query,

1733 leftmost_mapper,

1734 leftmost_attr,

1735 leftmost_relationship,

1736 entity,

1737 )

1738

1739 # generate another Query that will join the

1740 # left alias to the target relationships.

1741 # basically doing a longhand

1742 # "from_self()". (from_self() itself not quite industrial

1743 # strength enough for all contingencies...but very close)

1744

1745 q = query.Query(effective_entity)

1746

1747 q._execution_options = q._execution_options.union(

1748 {

1749 ("orig_query", SubqueryLoader): orig_query,

1750 ("subquery_paths", None): (subq_path, rewritten_path),

1751 }

1752 )

1753

1754 q = q._set_enable_single_crit(False)

1755 to_join, local_attr, parent_alias = self._prep_for_joins(

1756 left_alias, subq_path

1757 )

1758

1759 q = q.add_columns(*local_attr)

1760 q = self._apply_joins(

1761 q, to_join, left_alias, parent_alias, effective_entity

1762 )

1763

1764 q = self._setup_options(

1765 context,

1766 q,

1767 subq_path,

1768 rewritten_path,

1769 orig_query,

1770 effective_entity,

1771 loadopt,

1772 )

1773 q = self._setup_outermost_orderby(q)

1774

1775 return q

1776

1777 def create_row_processor(

1778 self,

1779 context,

1780 query_entity,

1781 path,

1782 loadopt,

1783 mapper,

1784 result,

1785 adapter,

1786 populators,

1787 ):

1788

1789 if context.refresh_state:

1790 return self._immediateload_create_row_processor(

1791 context,

1792 query_entity,

1793 path,

1794 loadopt,

1795 mapper,

1796 result,

1797 adapter,

1798 populators,

1799 )

1800 # the subqueryloader does a similar check in setup_query() unlike

1801 # the other post loaders, however we have this here for consistency

1802 elif self._check_recursive_postload(context, path, self.join_depth):

1803 return

1804 elif not isinstance(context.compile_state, ORMSelectCompileState):

1805 # issue 7505 - subqueryload() in 1.3 and previous would silently

1806 # degrade for from_statement() without warning. this behavior

1807 # is restored here

1808 return

1809

1810 if not self.parent.class_manager[self.key].impl.supports_population:

1811 raise sa_exc.InvalidRequestError(

1812 "'%s' does not support object "

1813 "population - eager loading cannot be applied." % self

1814 )

1815

1816 # a little dance here as the "path" is still something that only

1817 # semi-tracks the exact series of things we are loading, still not

1818 # telling us about with_polymorphic() and stuff like that when it's at

1819 # the root.. the initial MapperEntity is more accurate for this case.

1820 if len(path) == 1:

1821 if not orm_util._entity_isa(query_entity.entity_zero, self.parent):

1822 return

1823 elif not orm_util._entity_isa(path[-1], self.parent):

1824 return

1825

1826 subq = self._setup_query_from_rowproc(

1827 context,

1828 query_entity,

1829 path,

1830 path[-1],

1831 loadopt,

1832 adapter,

1833 )

1834

1835 if subq is None:

1836 return

1837

1838 assert subq.session is None

1839

1840 path = path[self.parent_property]

1841

1842 local_cols = self.parent_property.local_columns

1843

1844 # cache the loaded collections in the context

1845 # so that inheriting mappers don't re-load when they

1846 # call upon create_row_processor again

1847 collections = path.get(context.attributes, "collections")

1848 if collections is None:

1849 collections = self._SubqCollections(context, subq)

1850 path.set(context.attributes, "collections", collections)

1851

1852 if adapter:

1853 local_cols = [adapter.columns[c] for c in local_cols]

1854

1855 if self.uselist:

1856 self._create_collection_loader(

1857 context, result, collections, local_cols, populators

1858 )

1859 else:

1860 self._create_scalar_loader(

1861 context, result, collections, local_cols, populators

1862 )

1863

1864 def _create_collection_loader(

1865 self, context, result, collections, local_cols, populators

1866 ):

1867 tuple_getter = result._tuple_getter(local_cols)

1868

1869 def load_collection_from_subq(state, dict_, row):

1870 collection = collections.get(tuple_getter(row), ())

1871 state.get_impl(self.key).set_committed_value(

1872 state, dict_, collection

1873 )

1874

1875 def load_collection_from_subq_existing_row(state, dict_, row):

1876 if self.key not in dict_:

1877 load_collection_from_subq(state, dict_, row)

1878

1879 populators["new"].append((self.key, load_collection_from_subq))

1880 populators["existing"].append(

1881 (self.key, load_collection_from_subq_existing_row)

1882 )

1883

1884 if context.invoke_all_eagers:

1885 populators["eager"].append((self.key, collections.loader))

1886

1887 def _create_scalar_loader(

1888 self, context, result, collections, local_cols, populators

1889 ):

1890 tuple_getter = result._tuple_getter(local_cols)

1891

1892 def load_scalar_from_subq(state, dict_, row):

1893 collection = collections.get(tuple_getter(row), (None,))

1894 if len(collection) > 1:

1895 util.warn(

1896 "Multiple rows returned with "

1897 "uselist=False for eagerly-loaded attribute '%s' " % self

1898 )

1899

1900 scalar = collection[0]

1901 state.get_impl(self.key).set_committed_value(state, dict_, scalar)

1902

1903 def load_scalar_from_subq_existing_row(state, dict_, row):

1904 if self.key not in dict_:

1905 load_scalar_from_subq(state, dict_, row)

1906

1907 populators["new"].append((self.key, load_scalar_from_subq))

1908 populators["existing"].append(

1909 (self.key, load_scalar_from_subq_existing_row)

1910 )

1911 if context.invoke_all_eagers:

1912 populators["eager"].append((self.key, collections.loader))

1913

1914

1915@log.class_logger

1916@relationships.RelationshipProperty.strategy_for(lazy="joined")

1917@relationships.RelationshipProperty.strategy_for(lazy=False)

1918class JoinedLoader(AbstractRelationshipLoader):

1919 """Provide loading behavior for a :class:`.RelationshipProperty`

1920 using joined eager loading.

1921

1922 """

1923

1924 __slots__ = "join_depth", "_aliased_class_pool"

1925

1926 def __init__(self, parent, strategy_key):

1927 super(JoinedLoader, self).__init__(parent, strategy_key)

1928 self.join_depth = self.parent_property.join_depth

1929 self._aliased_class_pool = []

1930

1931 def init_class_attribute(self, mapper):

1932 self.parent_property._get_strategy(

1933 (("lazy", "select"),)

1934 ).init_class_attribute(mapper)

1935

1936 def setup_query(

1937 self,

1938 compile_state,

1939 query_entity,

1940 path,

1941 loadopt,

1942 adapter,

1943 column_collection=None,

1944 parentmapper=None,

1945 chained_from_outerjoin=False,

1946 **kwargs

1947 ):

1948 """Add a left outer join to the statement that's being constructed."""

1949

1950 if not compile_state.compile_options._enable_eagerloads:

1951 return

1952 elif self.uselist:

1953 compile_state.multi_row_eager_loaders = True

1954

1955 path = path[self.parent_property]

1956

1957 with_polymorphic = None

1958

1959 user_defined_adapter = (

1960 self._init_user_defined_eager_proc(

1961 loadopt, compile_state, compile_state.attributes

1962 )

1963 if loadopt

1964 else False

1965 )

1966

1967 if user_defined_adapter is not False:

1968

1969 # setup an adapter but dont create any JOIN, assume it's already

1970 # in the query

1971 (

1972 clauses,

1973 adapter,

1974 add_to_collection,

1975 ) = self._setup_query_on_user_defined_adapter(

1976 compile_state,

1977 query_entity,

1978 path,

1979 adapter,

1980 user_defined_adapter,

1981 )

1982

1983 # don't do "wrap" for multi-row, we want to wrap

1984 # limited/distinct SELECT,

1985 # because we want to put the JOIN on the outside.

1986

1987 else:

1988 # if not via query option, check for

1989 # a cycle

1990 if not path.contains(compile_state.attributes, "loader"):

1991 if self.join_depth:

1992 if path.length / 2 > self.join_depth:

1993 return

1994 elif path.contains_mapper(self.mapper):

1995 return

1996

1997 # add the JOIN and create an adapter

1998 (

1999 clauses,

2000 adapter,

2001 add_to_collection,

2002 chained_from_outerjoin,

2003 ) = self._generate_row_adapter(

2004 compile_state,

2005 query_entity,

2006 path,

2007 loadopt,

2008 adapter,

2009 column_collection,

2010 parentmapper,

2011 chained_from_outerjoin,

2012 )

2013

2014 # for multi-row, we want to wrap limited/distinct SELECT,

2015 # because we want to put the JOIN on the outside.

2016 compile_state.eager_adding_joins = True

2017

2018 with_poly_entity = path.get(

2019 compile_state.attributes, "path_with_polymorphic", None

2020 )

2021 if with_poly_entity is not None:

2022 with_polymorphic = inspect(

2023 with_poly_entity

2024 ).with_polymorphic_mappers

2025 else:

2026 with_polymorphic = None

2027

2028 path = path[self.entity]

2029

2030 loading._setup_entity_query(

2031 compile_state,

2032 self.mapper,

2033 query_entity,

2034 path,

2035 clauses,

2036 add_to_collection,

2037 with_polymorphic=with_polymorphic,

2038 parentmapper=self.mapper,

2039 chained_from_outerjoin=chained_from_outerjoin,

2040 )

2041

2042 if with_poly_entity is not None and None in set(

2043 compile_state.secondary_columns

2044 ):

2045 raise sa_exc.InvalidRequestError(

2046 "Detected unaliased columns when generating joined "

2047 "load. Make sure to use aliased=True or flat=True "

2048 "when using joined loading with with_polymorphic()."

2049 )

2050

2051 def _init_user_defined_eager_proc(

2052 self, loadopt, compile_state, target_attributes

2053 ):

2054

2055 # check if the opt applies at all

2056 if "eager_from_alias" not in loadopt.local_opts:

2057 # nope

2058 return False

2059

2060 path = loadopt.path.parent

2061

2062 # the option applies. check if the "user_defined_eager_row_processor"

2063 # has been built up.

2064 adapter = path.get(

2065 compile_state.attributes, "user_defined_eager_row_processor", False

2066 )

2067 if adapter is not False:

2068 # just return it

2069 return adapter

2070

2071 # otherwise figure it out.

2072 alias = loadopt.local_opts["eager_from_alias"]

2073 root_mapper, prop = path[-2:]

2074

2075 if alias is not None:

2076 if isinstance(alias, str):

2077 alias = prop.target.alias(alias)

2078 adapter = sql_util.ColumnAdapter(

2079 alias, equivalents=prop.mapper._equivalent_columns

2080 )

2081 else:

2082 if path.contains(

2083 compile_state.attributes, "path_with_polymorphic"

2084 ):

2085 with_poly_entity = path.get(

2086 compile_state.attributes, "path_with_polymorphic"

2087 )

2088 adapter = orm_util.ORMAdapter(

2089 with_poly_entity,

2090 equivalents=prop.mapper._equivalent_columns,

2091 )

2092 else:

2093 adapter = compile_state._polymorphic_adapters.get(

2094 prop.mapper, None

2095 )

2096 path.set(

2097 target_attributes,

2098 "user_defined_eager_row_processor",

2099 adapter,

2100 )

2101

2102 return adapter

2103

2104 def _setup_query_on_user_defined_adapter(

2105 self, context, entity, path, adapter, user_defined_adapter

2106 ):

2107

2108 # apply some more wrapping to the "user defined adapter"

2109 # if we are setting up the query for SQL render.

2110 adapter = entity._get_entity_clauses(context)

2111

2112 if adapter and user_defined_adapter:

2113 user_defined_adapter = user_defined_adapter.wrap(adapter)

2114 path.set(

2115 context.attributes,

2116 "user_defined_eager_row_processor",

2117 user_defined_adapter,

2118 )

2119 elif adapter:

2120 user_defined_adapter = adapter

2121 path.set(

2122 context.attributes,

2123 "user_defined_eager_row_processor",

2124 user_defined_adapter,

2125 )

2126

2127 add_to_collection = context.primary_columns

2128 return user_defined_adapter, adapter, add_to_collection

2129

2130 def _gen_pooled_aliased_class(self, context):

2131 # keep a local pool of AliasedClass objects that get re-used.

2132 # we need one unique AliasedClass per query per appearance of our

2133 # entity in the query.

2134

2135 if inspect(self.entity).is_aliased_class:

2136 alt_selectable = inspect(self.entity).selectable

2137 else:

2138 alt_selectable = None

2139

2140 key = ("joinedloader_ac", self)

2141 if key not in context.attributes:

2142 context.attributes[key] = idx = 0

2143 else:

2144 context.attributes[key] = idx = context.attributes[key] + 1

2145

2146 if idx >= len(self._aliased_class_pool):

2147 to_adapt = orm_util.AliasedClass(

2148 self.mapper,

2149 alias=alt_selectable._anonymous_fromclause(flat=True)

2150 if alt_selectable is not None

2151 else None,

2152 flat=True,

2153 use_mapper_path=True,

2154 )

2155

2156 # load up the .columns collection on the Alias() before

2157 # the object becomes shared among threads. this prevents

2158 # races for column identities.

2159 inspect(to_adapt).selectable.c

2160 self._aliased_class_pool.append(to_adapt)

2161

2162 return self._aliased_class_pool[idx]

2163

2164 def _generate_row_adapter(

2165 self,

2166 compile_state,

2167 entity,

2168 path,

2169 loadopt,

2170 adapter,

2171 column_collection,

2172 parentmapper,

2173 chained_from_outerjoin,

2174 ):

2175 with_poly_entity = path.get(

2176 compile_state.attributes, "path_with_polymorphic", None

2177 )

2178 if with_poly_entity:

2179 to_adapt = with_poly_entity

2180 else:

2181 to_adapt = self._gen_pooled_aliased_class(compile_state)

2182

2183 clauses = inspect(to_adapt)._memo(

2184 ("joinedloader_ormadapter", self),

2185 orm_util.ORMAdapter,

2186 to_adapt,

2187 equivalents=self.mapper._equivalent_columns,

2188 adapt_required=True,

2189 allow_label_resolve=False,

2190 anonymize_labels=True,

2191 )

2192

2193 assert clauses.aliased_class is not None

2194

2195 innerjoin = (

2196 loadopt.local_opts.get("innerjoin", self.parent_property.innerjoin)

2197 if loadopt is not None

2198 else self.parent_property.innerjoin

2199 )

2200

2201 if not innerjoin:

2202 # if this is an outer join, all non-nested eager joins from

2203 # this path must also be outer joins

2204 chained_from_outerjoin = True

2205

2206 compile_state.create_eager_joins.append(

2207 (

2208 self._create_eager_join,

2209 entity,

2210 path,

2211 adapter,

2212 parentmapper,

2213 clauses,

2214 innerjoin,

2215 chained_from_outerjoin,

2216 loadopt._extra_criteria if loadopt else (),

2217 )

2218 )

2219

2220 add_to_collection = compile_state.secondary_columns

2221 path.set(compile_state.attributes, "eager_row_processor", clauses)

2222

2223 return clauses, adapter, add_to_collection, chained_from_outerjoin

2224

2225 def _create_eager_join(

2226 self,

2227 compile_state,

2228 query_entity,

2229 path,

2230 adapter,

2231 parentmapper,

2232 clauses,

2233 innerjoin,

2234 chained_from_outerjoin,

2235 extra_criteria,

2236 ):

2237 if parentmapper is None:

2238 localparent = query_entity.mapper

2239 else:

2240 localparent = parentmapper

2241

2242 # whether or not the Query will wrap the selectable in a subquery,

2243 # and then attach eager load joins to that (i.e., in the case of

2244 # LIMIT/OFFSET etc.)

2245 should_nest_selectable = (

2246 compile_state.multi_row_eager_loaders

2247 and compile_state._should_nest_selectable

2248 )

2249

2250 query_entity_key = None

2251

2252 if (

2253 query_entity not in compile_state.eager_joins

2254 and not should_nest_selectable

2255 and compile_state.from_clauses

2256 ):

2257

2258 indexes = sql_util.find_left_clause_that_matches_given(

2259 compile_state.from_clauses, query_entity.selectable

2260 )

2261

2262 if len(indexes) > 1:

2263 # for the eager load case, I can't reproduce this right

2264 # now. For query.join() I can.

2265 raise sa_exc.InvalidRequestError(

2266 "Can't identify which query entity in which to joined "

2267 "eager load from. Please use an exact match when "

2268 "specifying the join path."

2269 )

2270

2271 if indexes:

2272 clause = compile_state.from_clauses[indexes[0]]

2273 # join to an existing FROM clause on the query.

2274 # key it to its list index in the eager_joins dict.

2275 # Query._compile_context will adapt as needed and

2276 # append to the FROM clause of the select().

2277 query_entity_key, default_towrap = indexes[0], clause

2278

2279 if query_entity_key is None:

2280 query_entity_key, default_towrap = (

2281 query_entity,

2282 query_entity.selectable,

2283 )

2284

2285 towrap = compile_state.eager_joins.setdefault(

2286 query_entity_key, default_towrap

2287 )

2288

2289 if adapter:

2290 if getattr(adapter, "aliased_class", None):

2291 # joining from an adapted entity. The adapted entity

2292 # might be a "with_polymorphic", so resolve that to our

2293 # specific mapper's entity before looking for our attribute

2294 # name on it.

2295 efm = inspect(adapter.aliased_class)._entity_for_mapper(

2296 localparent

2297 if localparent.isa(self.parent)

2298 else self.parent

2299 )

2300

2301 # look for our attribute on the adapted entity, else fall back

2302 # to our straight property

2303 onclause = getattr(efm.entity, self.key, self.parent_property)

2304 else:

2305 onclause = getattr(

2306 orm_util.AliasedClass(

2307 self.parent, adapter.selectable, use_mapper_path=True

2308 ),

2309 self.key,

2310 self.parent_property,

2311 )

2312

2313 else:

2314 onclause = self.parent_property

2315

2316 assert clauses.aliased_class is not None

2317

2318 attach_on_outside = (

2319 not chained_from_outerjoin

2320 or not innerjoin

2321 or innerjoin == "unnested"

2322 or query_entity.entity_zero.represents_outer_join

2323 )

2324

2325 extra_join_criteria = extra_criteria

2326 additional_entity_criteria = compile_state.global_attributes.get(

2327 ("additional_entity_criteria", self.mapper), ()

2328 )

2329 if additional_entity_criteria:

2330 extra_join_criteria += tuple(

2331 ae._resolve_where_criteria(self.mapper)

2332 for ae in additional_entity_criteria

2333 if ae.propagate_to_loaders

2334 )

2335

2336 if attach_on_outside:

2337 # this is the "classic" eager join case.

2338 eagerjoin = orm_util._ORMJoin(

2339 towrap,

2340 clauses.aliased_class,

2341 onclause,

2342 isouter=not innerjoin

2343 or query_entity.entity_zero.represents_outer_join

2344 or (chained_from_outerjoin and isinstance(towrap, sql.Join)),

2345 _left_memo=self.parent,

2346 _right_memo=self.mapper,

2347 _extra_criteria=extra_join_criteria,

2348 )

2349 else:

2350 # all other cases are innerjoin=='nested' approach

2351 eagerjoin = self._splice_nested_inner_join(

2352 path, towrap, clauses, onclause, extra_join_criteria

2353 )

2354

2355 compile_state.eager_joins[query_entity_key] = eagerjoin

2356

2357 # send a hint to the Query as to where it may "splice" this join

2358 eagerjoin.stop_on = query_entity.selectable

2359

2360 if not parentmapper:

2361 # for parentclause that is the non-eager end of the join,

2362 # ensure all the parent cols in the primaryjoin are actually

2363 # in the

2364 # columns clause (i.e. are not deferred), so that aliasing applied

2365 # by the Query propagates those columns outward.

2366 # This has the effect

2367 # of "undefering" those columns.

2368 for col in sql_util._find_columns(

2369 self.parent_property.primaryjoin

2370 ):

2371 if localparent.persist_selectable.c.contains_column(col):

2372 if adapter:

2373 col = adapter.columns[col]

2374 compile_state._append_dedupe_col_collection(

2375 col, compile_state.primary_columns

2376 )

2377

2378 if self.parent_property.order_by:

2379 compile_state.eager_order_by += tuple(

2380 (eagerjoin._target_adapter.copy_and_process)(

2381 util.to_list(self.parent_property.order_by)

2382 )

2383 )

2384

2385 def _splice_nested_inner_join(

2386 self, path, join_obj, clauses, onclause, extra_criteria, splicing=False

2387 ):

2388

2389 # recursive fn to splice a nested join into an existing one.

2390 # splicing=False means this is the outermost call, and it

2391 # should return a value. splicing=<from object> is the recursive

2392 # form, where it can return None to indicate the end of the recursion

2393

2394 if splicing is False:

2395 # first call is always handed a join object

2396 # from the outside

2397 assert isinstance(join_obj, orm_util._ORMJoin)

2398 elif isinstance(join_obj, sql.selectable.FromGrouping):

2399 return self._splice_nested_inner_join(

2400 path,

2401 join_obj.element,

2402 clauses,

2403 onclause,

2404 extra_criteria,

2405 splicing,

2406 )

2407 elif not isinstance(join_obj, orm_util._ORMJoin):

2408 if path[-2].isa(splicing):

2409 return orm_util._ORMJoin(

2410 join_obj,

2411 clauses.aliased_class,

2412 onclause,

2413 isouter=False,

2414 _left_memo=splicing,

2415 _right_memo=path[-1].mapper,

2416 _extra_criteria=extra_criteria,

2417 )

2418 else:

2419 return None

2420

2421 target_join = self._splice_nested_inner_join(

2422 path,

2423 join_obj.right,

2424 clauses,

2425 onclause,

2426 extra_criteria,

2427 join_obj._right_memo,

2428 )

2429 if target_join is None:

2430 right_splice = False

2431 target_join = self._splice_nested_inner_join(

2432 path,

2433 join_obj.left,

2434 clauses,

2435 onclause,

2436 extra_criteria,

2437 join_obj._left_memo,

2438 )

2439 if target_join is None:

2440 # should only return None when recursively called,

2441 # e.g. splicing refers to a from obj

2442 assert (

2443 splicing is not False

2444 ), "assertion failed attempting to produce joined eager loads"

2445 return None

2446 else:

2447 right_splice = True

2448

2449 if right_splice:

2450 # for a right splice, attempt to flatten out

2451 # a JOIN b JOIN c JOIN .. to avoid needless

2452 # parenthesis nesting

2453 if not join_obj.isouter and not target_join.isouter:

2454 eagerjoin = join_obj._splice_into_center(target_join)

2455 else:

2456 eagerjoin = orm_util._ORMJoin(

2457 join_obj.left,

2458 target_join,

2459 join_obj.onclause,

2460 isouter=join_obj.isouter,

2461 _left_memo=join_obj._left_memo,

2462 )

2463 else:

2464 eagerjoin = orm_util._ORMJoin(

2465 target_join,

2466 join_obj.right,

2467 join_obj.onclause,

2468 isouter=join_obj.isouter,

2469 _right_memo=join_obj._right_memo,

2470 )

2471

2472 eagerjoin._target_adapter = target_join._target_adapter

2473 return eagerjoin

2474

2475 def _create_eager_adapter(self, context, result, adapter, path, loadopt):

2476 compile_state = context.compile_state

2477

2478 user_defined_adapter = (

2479 self._init_user_defined_eager_proc(

2480 loadopt, compile_state, context.attributes

2481 )

2482 if loadopt

2483 else False

2484 )

2485

2486 if user_defined_adapter is not False:

2487 decorator = user_defined_adapter

2488 # user defined eagerloads are part of the "primary"

2489 # portion of the load.

2490 # the adapters applied to the Query should be honored.

2491 if compile_state.compound_eager_adapter and decorator:

2492 decorator = decorator.wrap(

2493 compile_state.compound_eager_adapter

2494 )

2495 elif compile_state.compound_eager_adapter:

2496 decorator = compile_state.compound_eager_adapter

2497 else:

2498 decorator = path.get(

2499 compile_state.attributes, "eager_row_processor"

2500 )

2501 if decorator is None:

2502 return False

2503

2504 if self.mapper._result_has_identity_key(result, decorator):

2505 return decorator

2506 else:

2507 # no identity key - don't return a row

2508 # processor, will cause a degrade to lazy

2509 return False

2510

2511 def create_row_processor(

2512 self,

2513 context,

2514 query_entity,

2515 path,

2516 loadopt,

2517 mapper,

2518 result,

2519 adapter,

2520 populators,

2521 ):

2522 if not self.parent.class_manager[self.key].impl.supports_population:

2523 raise sa_exc.InvalidRequestError(

2524 "'%s' does not support object "

2525 "population - eager loading cannot be applied." % self

2526 )

2527

2528 if self.uselist:

2529 context.loaders_require_uniquing = True

2530

2531 our_path = path[self.parent_property]

2532

2533 eager_adapter = self._create_eager_adapter(

2534 context, result, adapter, our_path, loadopt

2535 )

2536

2537 if eager_adapter is not False:

2538 key = self.key

2539

2540 _instance = loading._instance_processor(

2541 query_entity,

2542 self.mapper,

2543 context,

2544 result,

2545 our_path[self.entity],

2546 eager_adapter,

2547 )

2548

2549 if not self.uselist:

2550 self._create_scalar_loader(context, key, _instance, populators)

2551 else:

2552 self._create_collection_loader(

2553 context, key, _instance, populators

2554 )

2555 else:

2556 self.parent_property._get_strategy(

2557 (("lazy", "select"),)

2558 ).create_row_processor(

2559 context,

2560 query_entity,

2561 path,

2562 loadopt,

2563 mapper,

2564 result,

2565 adapter,

2566 populators,

2567 )

2568

2569 def _create_collection_loader(self, context, key, _instance, populators):

2570 def load_collection_from_joined_new_row(state, dict_, row):

2571 # note this must unconditionally clear out any existing collection.

2572 # an existing collection would be present only in the case of

2573 # populate_existing().

2574 collection = attributes.init_state_collection(state, dict_, key)

2575 result_list = util.UniqueAppender(

2576 collection, "append_without_event"

2577 )

2578 context.attributes[(state, key)] = result_list

2579 inst = _instance(row)

2580 if inst is not None:

2581 result_list.append(inst)

2582

2583 def load_collection_from_joined_existing_row(state, dict_, row):

2584 if (state, key) in context.attributes:

2585 result_list = context.attributes[(state, key)]

2586 else:

2587 # appender_key can be absent from context.attributes

2588 # with isnew=False when self-referential eager loading

2589 # is used; the same instance may be present in two

2590 # distinct sets of result columns

2591 collection = attributes.init_state_collection(

2592 state, dict_, key

2593 )

2594 result_list = util.UniqueAppender(

2595 collection, "append_without_event"

2596 )

2597 context.attributes[(state, key)] = result_list

2598 inst = _instance(row)

2599 if inst is not None:

2600 result_list.append(inst)

2601

2602 def load_collection_from_joined_exec(state, dict_, row):

2603 _instance(row)

2604

2605 populators["new"].append(

2606 (self.key, load_collection_from_joined_new_row)

2607 )

2608 populators["existing"].append(

2609 (self.key, load_collection_from_joined_existing_row)

2610 )

2611 if context.invoke_all_eagers:

2612 populators["eager"].append(

2613 (self.key, load_collection_from_joined_exec)

2614 )

2615

2616 def _create_scalar_loader(self, context, key, _instance, populators):

2617 def load_scalar_from_joined_new_row(state, dict_, row):

2618 # set a scalar object instance directly on the parent

2619 # object, bypassing InstrumentedAttribute event handlers.

2620 dict_[key] = _instance(row)

2621

2622 def load_scalar_from_joined_existing_row(state, dict_, row):

2623 # call _instance on the row, even though the object has

2624 # been created, so that we further descend into properties

2625 existing = _instance(row)

2626

2627 # conflicting value already loaded, this shouldn't happen

2628 if key in dict_:

2629 if existing is not dict_[key]:

2630 util.warn(

2631 "Multiple rows returned with "

2632 "uselist=False for eagerly-loaded attribute '%s' "

2633 % self

2634 )

2635 else:

2636 # this case is when one row has multiple loads of the

2637 # same entity (e.g. via aliasing), one has an attribute

2638 # that the other doesn't.

2639 dict_[key] = existing

2640

2641 def load_scalar_from_joined_exec(state, dict_, row):

2642 _instance(row)

2643

2644 populators["new"].append((self.key, load_scalar_from_joined_new_row))

2645 populators["existing"].append(

2646 (self.key, load_scalar_from_joined_existing_row)

2647 )

2648 if context.invoke_all_eagers:

2649 populators["eager"].append(

2650 (self.key, load_scalar_from_joined_exec)

2651 )

2652

2653

2654@log.class_logger

2655@relationships.RelationshipProperty.strategy_for(lazy="selectin")

2656class SelectInLoader(PostLoader, util.MemoizedSlots):

2657 __slots__ = (

2658 "join_depth",

2659 "omit_join",

2660 "_parent_alias",

2661 "_query_info",

2662 "_fallback_query_info",

2663 )

2664

2665 query_info = collections.namedtuple(

2666 "queryinfo",

2667 [

2668 "load_only_child",

2669 "load_with_join",

2670 "in_expr",

2671 "pk_cols",

2672 "zero_idx",

2673 "child_lookup_cols",

2674 ],

2675 )

2676

2677 _chunksize = 500

2678

2679 def __init__(self, parent, strategy_key):

2680 super(SelectInLoader, self).__init__(parent, strategy_key)

2681 self.join_depth = self.parent_property.join_depth

2682 is_m2o = self.parent_property.direction is interfaces.MANYTOONE

2683

2684 if self.parent_property.omit_join is not None:

2685 self.omit_join = self.parent_property.omit_join

2686 else:

2687 lazyloader = self.parent_property._get_strategy(

2688 (("lazy", "select"),)

2689 )

2690 if is_m2o:

2691 self.omit_join = lazyloader.use_get

2692 else:

2693 self.omit_join = self.parent._get_clause[0].compare(

2694 lazyloader._rev_lazywhere,

2695 use_proxies=True,

2696 compare_keys=False,

2697 equivalents=self.parent._equivalent_columns,

2698 )

2699

2700 if self.omit_join:

2701 if is_m2o:

2702 self._query_info = self._init_for_omit_join_m2o()

2703 self._fallback_query_info = self._init_for_join()

2704 else:

2705 self._query_info = self._init_for_omit_join()

2706 else:

2707 self._query_info = self._init_for_join()

2708

2709 def _init_for_omit_join(self):

2710 pk_to_fk = dict(

2711 self.parent_property._join_condition.local_remote_pairs

2712 )

2713 pk_to_fk.update(

2714 (equiv, pk_to_fk[k])

2715 for k in list(pk_to_fk)

2716 for equiv in self.parent._equivalent_columns.get(k, ())

2717 )

2718

2719 pk_cols = fk_cols = [

2720 pk_to_fk[col] for col in self.parent.primary_key if col in pk_to_fk

2721 ]

2722 if len(fk_cols) > 1:

2723 in_expr = sql.tuple_(*fk_cols)

2724 zero_idx = False

2725 else:

2726 in_expr = fk_cols[0]

2727 zero_idx = True

2728

2729 return self.query_info(False, False, in_expr, pk_cols, zero_idx, None)

2730

2731 def _init_for_omit_join_m2o(self):

2732 pk_cols = self.mapper.primary_key

2733 if len(pk_cols) > 1:

2734 in_expr = sql.tuple_(*pk_cols)

2735 zero_idx = False

2736 else:

2737 in_expr = pk_cols[0]

2738 zero_idx = True

2739

2740 lazyloader = self.parent_property._get_strategy((("lazy", "select"),))

2741 lookup_cols = [lazyloader._equated_columns[pk] for pk in pk_cols]

2742

2743 return self.query_info(

2744 True, False, in_expr, pk_cols, zero_idx, lookup_cols

2745 )

2746

2747 def _init_for_join(self):

2748 self._parent_alias = aliased(self.parent.class_)

2749 pa_insp = inspect(self._parent_alias)

2750 pk_cols = [

2751 pa_insp._adapt_element(col) for col in self.parent.primary_key

2752 ]

2753 if len(pk_cols) > 1:

2754 in_expr = sql.tuple_(*pk_cols)

2755 zero_idx = False

2756 else:

2757 in_expr = pk_cols[0]

2758 zero_idx = True

2759 return self.query_info(False, True, in_expr, pk_cols, zero_idx, None)

2760

2761 def init_class_attribute(self, mapper):

2762 self.parent_property._get_strategy(

2763 (("lazy", "select"),)

2764 ).init_class_attribute(mapper)

2765

2766 def create_row_processor(

2767 self,

2768 context,

2769 query_entity,

2770 path,

2771 loadopt,

2772 mapper,

2773 result,

2774 adapter,

2775 populators,

2776 ):

2777

2778 if context.refresh_state:

2779 return self._immediateload_create_row_processor(

2780 context,

2781 query_entity,

2782 path,

2783 loadopt,

2784 mapper,

2785 result,

2786 adapter,

2787 populators,

2788 )

2789 elif self._check_recursive_postload(context, path, self.join_depth):

2790 return

2791

2792 if not self.parent.class_manager[self.key].impl.supports_population:

2793 raise sa_exc.InvalidRequestError(

2794 "'%s' does not support object "

2795 "population - eager loading cannot be applied." % self

2796 )

2797

2798 # a little dance here as the "path" is still something that only

2799 # semi-tracks the exact series of things we are loading, still not

2800 # telling us about with_polymorphic() and stuff like that when it's at

2801 # the root.. the initial MapperEntity is more accurate for this case.

2802 if len(path) == 1:

2803 if not orm_util._entity_isa(query_entity.entity_zero, self.parent):

2804 return

2805 elif not orm_util._entity_isa(path[-1], self.parent):

2806 return

2807

2808 selectin_path = (

2809 context.compile_state.current_path or orm_util.PathRegistry.root

2810 ) + path

2811

2812 path_w_prop = path[self.parent_property]

2813

2814 # build up a path indicating the path from the leftmost

2815 # entity to the thing we're subquery loading.

2816 with_poly_entity = path_w_prop.get(

2817 context.attributes, "path_with_polymorphic", None

2818 )

2819 if with_poly_entity is not None:

2820 effective_entity = inspect(with_poly_entity)

2821 else:

2822 effective_entity = self.entity

2823

2824 loading.PostLoad.callable_for_path(

2825 context,

2826 selectin_path,

2827 self.parent,

2828 self.parent_property,

2829 self._load_for_path,

2830 effective_entity,

2831 loadopt,

2832 )

2833

2834 def _load_for_path(

2835 self, context, path, states, load_only, effective_entity, loadopt

2836 ):

2837 if load_only and self.key not in load_only:

2838 return

2839

2840 query_info = self._query_info

2841

2842 if query_info.load_only_child:

2843 our_states = collections.defaultdict(list)

2844 none_states = []

2845

2846 mapper = self.parent

2847

2848 for state, overwrite in states:

2849 state_dict = state.dict

2850 related_ident = tuple(

2851 mapper._get_state_attr_by_column(

2852 state,

2853 state_dict,

2854 lk,

2855 passive=attributes.PASSIVE_NO_FETCH,

2856 )

2857 for lk in query_info.child_lookup_cols

2858 )

2859 # if the loaded parent objects do not have the foreign key

2860 # to the related item loaded, then degrade into the joined

2861 # version of selectinload

2862 if attributes.PASSIVE_NO_RESULT in related_ident:

2863 query_info = self._fallback_query_info

2864 break

2865

2866 # organize states into lists keyed to particular foreign

2867 # key values.

2868 if None not in related_ident:

2869 our_states[related_ident].append(

2870 (state, state_dict, overwrite)

2871 )

2872 else:

2873 # For FK values that have None, add them to a

2874 # separate collection that will be populated separately

2875 none_states.append((state, state_dict, overwrite))

2876

2877 # note the above conditional may have changed query_info

2878 if not query_info.load_only_child:

2879 our_states = [

2880 (state.key[1], state, state.dict, overwrite)

2881 for state, overwrite in states

2882 ]

2883

2884 pk_cols = query_info.pk_cols

2885 in_expr = query_info.in_expr

2886

2887 if not query_info.load_with_join:

2888 # in "omit join" mode, the primary key column and the

2889 # "in" expression are in terms of the related entity. So

2890 # if the related entity is polymorphic or otherwise aliased,

2891 # we need to adapt our "pk_cols" and "in_expr" to that

2892 # entity. in non-"omit join" mode, these are against the

2893 # parent entity and do not need adaption.

2894 if effective_entity.is_aliased_class:

2895 pk_cols = [

2896 effective_entity._adapt_element(col) for col in pk_cols

2897 ]

2898 in_expr = effective_entity._adapt_element(in_expr)

2899

2900 bundle_ent = orm_util.Bundle("pk", *pk_cols)

2901 bundle_sql = bundle_ent.__clause_element__()

2902

2903 entity_sql = effective_entity.__clause_element__()

2904 q = Select._create_raw_select(

2905 _raw_columns=[bundle_sql, entity_sql],

2906 _label_style=LABEL_STYLE_TABLENAME_PLUS_COL,

2907 _compile_options=ORMCompileState.default_compile_options,

2908 _propagate_attrs={

2909 "compile_state_plugin": "orm",

2910 "plugin_subject": effective_entity,

2911 },

2912 )

2913

2914 if not query_info.load_with_join:

2915 # the Bundle we have in the "omit_join" case is against raw, non

2916 # annotated columns, so to ensure the Query knows its primary

2917 # entity, we add it explicitly. If we made the Bundle against

2918 # annotated columns, we hit a performance issue in this specific

2919 # case, which is detailed in issue #4347.

2920 q = q.select_from(effective_entity)

2921 else:

2922 # in the non-omit_join case, the Bundle is against the annotated/

2923 # mapped column of the parent entity, but the #4347 issue does not

2924 # occur in this case.

2925 q = q.select_from(self._parent_alias).join(

2926 getattr(self._parent_alias, self.parent_property.key).of_type(

2927 effective_entity

2928 )

2929 )

2930

2931 q = q.filter(in_expr.in_(sql.bindparam("primary_keys")))

2932

2933 # a test which exercises what these comments talk about is

2934 # test_selectin_relations.py -> test_twolevel_selectin_w_polymorphic

2935 #

2936 # effective_entity above is given to us in terms of the cached

2937 # statement, namely this one:

2938 orig_query = context.compile_state.select_statement

2939

2940 # the actual statement that was requested is this one:

2941 # context_query = context.query

2942 #

2943 # that's not the cached one, however. So while it is of the identical

2944 # structure, if it has entities like AliasedInsp, which we get from

2945 # aliased() or with_polymorphic(), the AliasedInsp will likely be a

2946 # different object identity each time, and will not match up

2947 # hashing-wise to the corresponding AliasedInsp that's in the

2948 # cached query, meaning it won't match on paths and loader lookups

2949 # and loaders like this one will be skipped if it is used in options.

2950 #

2951 # as it turns out, standard loader options like selectinload(),

2952 # lazyload() that have a path need

2953 # to come from the cached query so that the AliasedInsp etc. objects

2954 # that are in the query line up with the object that's in the path

2955 # of the strategy object. however other options like

2956 # with_loader_criteria() that doesn't have a path (has a fixed entity)

2957 # and needs to have access to the latest closure state in order to

2958 # be correct, we need to use the uncached one.

2959 #

2960 # as of #8399 we let the loader option itself figure out what it

2961 # wants to do given cached and uncached version of itself.

2962

2963 effective_path = path[self.parent_property]

2964

2965 if orig_query is context.query:

2966 new_options = orig_query._with_options

2967 else:

2968 cached_options = orig_query._with_options

2969 uncached_options = context.query._with_options

2970

2971 # propagate compile state options from the original query,

2972 # updating their "extra_criteria" as necessary.

2973 # note this will create a different cache key than

2974 # "orig" options if extra_criteria is present, because the copy

2975 # of extra_criteria will have different boundparam than that of

2976 # the QueryableAttribute in the path

2977 new_options = [

2978 orig_opt._adapt_cached_option_to_uncached_option(

2979 context, uncached_opt

2980 )

2981 for orig_opt, uncached_opt in zip(

2982 cached_options, uncached_options

2983 )

2984 ]

2985

2986 if loadopt and loadopt._extra_criteria:

2987 new_options += (

2988 orm_util.LoaderCriteriaOption(

2989 effective_entity,

2990 loadopt._generate_extra_criteria(context),

2991 ),

2992 )

2993

2994 q = q.options(*new_options)

2995

2996 q = q._update_compile_options({"_current_path": effective_path})

2997 if context.populate_existing:

2998 q = q.execution_options(populate_existing=True)

2999

3000 if self.parent_property.order_by:

3001 if not query_info.load_with_join:

3002 eager_order_by = self.parent_property.order_by

3003 if effective_entity.is_aliased_class:

3004 eager_order_by = [

3005 effective_entity._adapt_element(elem)

3006 for elem in eager_order_by

3007 ]

3008 q = q.order_by(*eager_order_by)

3009 else:

3010

3011 def _setup_outermost_orderby(compile_context):

3012 compile_context.eager_order_by += tuple(

3013 util.to_list(self.parent_property.order_by)

3014 )

3015

3016 q = q._add_context_option(

3017 _setup_outermost_orderby, self.parent_property

3018 )

3019

3020 if query_info.load_only_child:

3021 self._load_via_child(

3022 our_states, none_states, query_info, q, context

3023 )

3024 else:

3025 self._load_via_parent(our_states, query_info, q, context)

3026

3027 def _load_via_child(self, our_states, none_states, query_info, q, context):

3028 uselist = self.uselist

3029

3030 # this sort is really for the benefit of the unit tests

3031 our_keys = sorted(our_states)

3032 while our_keys:

3033 chunk = our_keys[0 : self._chunksize]

3034 our_keys = our_keys[self._chunksize :]

3035 data = {

3036 k: v

3037 for k, v in context.session.execute(

3038 q,

3039 params={

3040 "primary_keys": [

3041 key[0] if query_info.zero_idx else key

3042 for key in chunk

3043 ]

3044 },

3045 ).unique()

3046 }

3047

3048 for key in chunk:

3049 # for a real foreign key and no concurrent changes to the

3050 # DB while running this method, "key" is always present in

3051 # data. However, for primaryjoins without real foreign keys

3052 # a non-None primaryjoin condition may still refer to no

3053 # related object.

3054 related_obj = data.get(key, None)

3055 for state, dict_, overwrite in our_states[key]:

3056 if not overwrite and self.key in dict_:

3057 continue

3058

3059 state.get_impl(self.key).set_committed_value(

3060 state,

3061 dict_,

3062 related_obj if not uselist else [related_obj],

3063 )

3064 # populate none states with empty value / collection

3065 for state, dict_, overwrite in none_states:

3066 if not overwrite and self.key in dict_:

3067 continue

3068

3069 # note it's OK if this is a uselist=True attribute, the empty

3070 # collection will be populated

3071 state.get_impl(self.key).set_committed_value(state, dict_, None)

3072

3073 def _load_via_parent(self, our_states, query_info, q, context):

3074 uselist = self.uselist

3075 _empty_result = () if uselist else None

3076

3077 while our_states:

3078 chunk = our_states[0 : self._chunksize]

3079 our_states = our_states[self._chunksize :]

3080

3081 primary_keys = [

3082 key[0] if query_info.zero_idx else key

3083 for key, state, state_dict, overwrite in chunk

3084 ]

3085

3086 data = collections.defaultdict(list)

3087 for k, v in itertools.groupby(

3088 context.session.execute(

3089 q, params={"primary_keys": primary_keys}

3090 ).unique(),

3091 lambda x: x[0],

3092 ):

3093 data[k].extend(vv[1] for vv in v)

3094

3095 for key, state, state_dict, overwrite in chunk:

3096

3097 if not overwrite and self.key in state_dict:

3098 continue

3099

3100 collection = data.get(key, _empty_result)

3101

3102 if not uselist and collection:

3103 if len(collection) > 1:

3104 util.warn(

3105 "Multiple rows returned with "

3106 "uselist=False for eagerly-loaded "

3107 "attribute '%s' " % self

3108 )

3109 state.get_impl(self.key).set_committed_value(

3110 state, state_dict, collection[0]

3111 )

3112 else:

3113 # note that empty tuple set on uselist=False sets the

3114 # value to None

3115 state.get_impl(self.key).set_committed_value(

3116 state, state_dict, collection

3117 )

3118

3119

3120def single_parent_validator(desc, prop):

3121 def _do_check(state, value, oldvalue, initiator):

3122 if value is not None and initiator.key == prop.key:

3123 hasparent = initiator.hasparent(attributes.instance_state(value))

3124 if hasparent and oldvalue is not value:

3125 raise sa_exc.InvalidRequestError(

3126 "Instance %s is already associated with an instance "

3127 "of %s via its %s attribute, and is only allowed a "

3128 "single parent."

3129 % (orm_util.instance_str(value), state.class_, prop),

3130 code="bbf1",

3131 )

3132 return value

3133

3134 def append(state, value, initiator):

3135 return _do_check(state, value, None, initiator)

3136

3137 def set_(state, value, oldvalue, initiator):

3138 return _do_check(state, value, oldvalue, initiator)

3139

3140 event.listen(

3141 desc, "append", append, raw=True, retval=True, active_history=True

3142 )

3143 event.listen(desc, "set", set_, raw=True, retval=True, active_history=True)