_compile_state_cls: Optional[Type[ORMCompileState]]
_starting_event_idx: int
_events_todo: List[Any]
- _update_execution_options: Optional[_ExecuteOptions]
+ _update_execution_options: _ExecuteOptions
def __init__(
self,
self.bind_arguments = bind_arguments
self._compile_state_cls = compile_state_cls
self._events_todo = list(events_todo)
+ self._update_execution_options = util.EMPTY_DICT
def _remaining_events(self) -> List[_InstanceLevelDispatch[Session]]:
return self._events_todo[self._starting_event_idx + 1 :]
def update_execution_options(self, **opts: Any) -> None:
"""Update the local execution options with new values."""
self.local_execution_options = self.local_execution_options.union(opts)
+ self._update_execution_options = self._update_execution_options.union(
+ opts
+ )
def _orm_compile_options(
self,
events_todo = list(events_todo) + [_add_event]
if events_todo:
+ # save the original execution options before
+ # orm_pre_session_exec processes them, so that we can pass
+ # the unprocessed options (plus any explicit updates from event
+ # hooks) to the second orm_pre_session_exec call. This
+ # prevents internal state like _sa_orm_load_options and
+ # yield_per from the first call leaking into the second call,
+ # which would otherwise cause issues like yield_per incorrectly
+ # propagating into post-load (selectinload etc.) queries.
+ # part of #13301.
+ original_execution_options = execution_options
+
if compile_state_cls is not None:
# for event handlers, do the orm_pre_session_exec
# pass ahead of the event handlers, so that things like
return fn_result
statement = orm_exec_state.statement
- execution_options = orm_exec_state.local_execution_options
+
+ # use the original execution options plus only the explicit
+ # updates from event hooks, not the processed options from
+ # the first orm_pre_session_exec call
+ execution_options = original_execution_options.union(
+ orm_exec_state._update_execution_options
+ )
if compile_state_cls is not None:
# now run orm_pre_session_exec() "for real". if there were
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import deferred
from sqlalchemy.orm import EXT_SKIP
+from sqlalchemy.orm import immediateload
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import lazyload
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import UserDefinedOption
+from sqlalchemy.orm.context import QueryContext
from sqlalchemy.sql.cache_key import NO_CACHE
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
+from sqlalchemy.testing.fixtures import RemovesEvents
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
from test.orm import _fixtures
-class ORMExecuteTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
+class ORMExecuteTest(
+ RemovesEvents, RemoveORMEventsGlobally, _fixtures.FixtureTest
+):
run_setup_mappers = "once"
run_inserts = "once"
run_deletes = None
assert False
if use_query_cache:
+ testing.db._compiled_cache.clear()
s = fixture_session()
else:
s = fixture_session(
),
)
+ @testing.combinations(
+ (lazyload,),
+ (selectinload,),
+ (immediateload,),
+ (joinedload,),
+ (subqueryload,),
+ argnames="loader_opt",
+ )
+ @testing.variation("use_event", [True, False])
+ def test_execution_option_visibility_and_mutation(
+ self, loader_opt, use_event
+ ):
+ """test the full contract of how options are passed to
+ do_orm_execute and then onto a relationship loader.
+
+ part of #13301
+
+ the non-event part of this test has overlap with
+ test_session.py -> test_eagerloader_exec_option, which is also
+ looking at how top level options find their way into eager loader
+ queries. in this test though we try to establish every possible
+ path for options with all loaders.
+
+ """
+ User, Address, Dinagling = self.classes("User", "Address", "Dingaling")
+
+ s = fixture_session()
+
+ m1 = Mock()
+
+ def _before_cursor_execute(
+ conn, cursor, statement, parameters, context, executemany
+ ):
+ opts = {
+ k: v
+ for k, v in context.execution_options.items()
+ if isinstance(k, str) and k.startswith("user_")
+ }
+ m1.before_cursor_execute(opts)
+
+ self.event_listen(
+ testing.db, "before_cursor_execute", _before_cursor_execute
+ )
+
+ if use_event:
+
+ def go(context):
+ opts = {
+ k: v
+ for k, v in context.execution_options.items()
+ if isinstance(k, str) and k.startswith("user_")
+ }
+ m1.do_orm_execute(context.is_relationship_load, opts)
+
+ context.update_execution_options(
+ user_defined_three="z",
+ user_defined_two_override="z2",
+ )
+
+ # local_execution_options receive the updates
+ eq_(
+ context.local_execution_options["user_defined_three"],
+ "z",
+ )
+ eq_(
+ context.local_execution_options[
+ "user_defined_two_override"
+ ],
+ "z2",
+ )
+
+ # the .execution_options does not show the updates
+ # (not sure if i like this but this is how it was
+ # done)
+ if not context.is_relationship_load:
+ # only test for the top-level load. the options
+ # may be present in a relationship load.
+ assert (
+ "user_defined_three" not in context.execution_options
+ )
+ eq_(
+ context.execution_options["user_defined_two_override"],
+ "y2",
+ )
+
+ self.event_listen(s, "do_orm_execute", go)
+
+ stmt = (
+ select(User)
+ .options(
+ loader_opt(User.addresses).options(
+ loader_opt(Address.dingaling)
+ ),
+ )
+ .execution_options(
+ user_defined_one="x",
+ user_defined_one_override="x2",
+ )
+ )
+
+ for u in (
+ s.execute(
+ stmt,
+ execution_options={
+ "user_defined_two": "y",
+ "user_defined_two_override": "y2",
+ "user_defined_one_override": "y2",
+ },
+ )
+ .unique()
+ .scalars()
+ ):
+ for a in u.addresses:
+ a.dingaling
+
+ if use_event:
+ upfront_expected_calls = [
+ call.do_orm_execute(
+ False,
+ {
+ "user_defined_one": "x",
+ "user_defined_two": "y",
+ "user_defined_two_override": "y2",
+ "user_defined_one_override": "y2",
+ },
+ ),
+ call.before_cursor_execute(
+ {
+ "user_defined_one": "x",
+ "user_defined_two_override": "z2",
+ "user_defined_one_override": "y2",
+ "user_defined_two": "y",
+ "user_defined_three": "z",
+ },
+ ),
+ ]
+ relationship_level_calls = [
+ call.do_orm_execute(
+ True,
+ {
+ "user_defined_one": "x",
+ "user_defined_two_override": "z2",
+ "user_defined_one_override": "y2",
+ "user_defined_two": "y",
+ "user_defined_three": "z",
+ },
+ ),
+ call.before_cursor_execute(
+ {
+ "user_defined_one": "x",
+ "user_defined_two_override": "z2",
+ "user_defined_one_override": "y2",
+ "user_defined_two": "y",
+ "user_defined_three": "z",
+ },
+ ),
+ ]
+ joined_and_lazy_calls = [
+ call.do_orm_execute(
+ True,
+ {},
+ ),
+ call.before_cursor_execute(
+ {
+ "user_defined_two_override": "z2",
+ "user_defined_three": "z",
+ },
+ ),
+ ]
+ else:
+ upfront_expected_calls = [
+ call.before_cursor_execute(
+ {
+ "user_defined_one": "x",
+ "user_defined_two_override": "y2",
+ "user_defined_one_override": "y2",
+ "user_defined_two": "y",
+ },
+ ),
+ ]
+ relationship_level_calls = [
+ call.before_cursor_execute(
+ {
+ "user_defined_one": "x",
+ "user_defined_two_override": "y2",
+ "user_defined_one_override": "y2",
+ "user_defined_two": "y",
+ },
+ ),
+ ]
+ joined_and_lazy_calls = [
+ call.before_cursor_execute({}),
+ ]
+
+ if loader_opt in (selectinload, subqueryload, immediateload):
+ eq_(
+ m1.mock_calls,
+ upfront_expected_calls
+ + (
+ relationship_level_calls
+ * (2 if loader_opt in (selectinload, subqueryload) else 9)
+ ),
+ )
+ elif loader_opt in (lazyload, joinedload):
+ eq_(
+ m1.mock_calls,
+ upfront_expected_calls
+ + (
+ joined_and_lazy_calls
+ * (9 if loader_opt is lazyload else 0)
+ ),
+ )
+
+ @testing.combinations(
+ (selectinload,),
+ (immediateload,),
+ argnames="loader_opt",
+ )
+ @testing.variation("option_type", ["yield_per", "autoflush"])
+ @testing.variation(
+ "event_style",
+ [
+ "no_event",
+ "noop_event",
+ "set_in_event_for_relationship",
+ ],
+ )
+ def test_load_option_propagation_to_post_load(
+ self,
+ loader_opt,
+ option_type,
+ event_style,
+ ):
+ """test that yield_per and autoflush options set on the
+ top-level query don't leak into load_options for post-load
+ queries (selectinload, immediateload), and that options
+ explicitly set via update_execution_options in a
+ do_orm_execute event hook for relationship loads do propagate
+ correctly.
+
+ part of #13301
+
+ """
+
+ User = self.classes.User
+
+ s = fixture_session()
+
+ expected_lead_options = {
+ "yield_per": None,
+ "autoflush": True,
+ "is_post_load": False,
+ }
+ expected_post_load_options = {
+ "yield_per": None,
+ "autoflush": True,
+ "is_post_load": True,
+ }
+ operation_should_fail = (
+ event_style.set_in_event_for_relationship and option_type.yield_per
+ )
+
+ if event_style.noop_event:
+
+ def go(context):
+ pass
+
+ self.event_listen(s, "do_orm_execute", go)
+
+ elif event_style.set_in_event_for_relationship:
+
+ def go(context):
+ if context.is_relationship_load:
+ if option_type.yield_per:
+ context.update_execution_options(yield_per=7)
+ expected_post_load_options["yield_per"] = 7
+ elif option_type.autoflush:
+ context.update_execution_options(autoflush=False)
+ expected_post_load_options["autoflush"] = False
+
+ self.event_listen(s, "do_orm_execute", go)
+ elif event_style.no_event:
+ pass
+
+ captured_load_options = []
+
+ orig_init = QueryContext.__init__
+
+ def tracking_init(self_qc, *args, **kwargs):
+ orig_init(self_qc, *args, **kwargs)
+ captured_load_options.append(
+ {
+ "yield_per": self_qc.load_options._yield_per,
+ "autoflush": (self_qc.load_options._autoflush),
+ "is_post_load": (
+ self_qc.load_options._sa_top_level_orm_context
+ is not None
+ ),
+ }
+ )
+
+ with patch.object(QueryContext, "__init__", tracking_init):
+
+ stmt = select(User).options(loader_opt(User.addresses))
+
+ if not event_style.set_in_event_for_relationship:
+ if option_type.yield_per:
+ stmt = stmt.execution_options(yield_per=5)
+ expected_lead_options["yield_per"] = 5
+ elif option_type.autoflush:
+ stmt = stmt.execution_options(autoflush=False)
+ expected_lead_options["autoflush"] = False
+
+ if operation_should_fail:
+ with expect_raises_message(
+ sa_exc.InvalidRequestError,
+ "Can't use the ORM yield_per feature in "
+ "conjunction with unique",
+ ):
+ s.scalars(stmt).all()
+ else:
+ s.scalars(stmt).all()
+
+ eq_(
+ captured_load_options,
+ [expected_lead_options]
+ + (
+ [expected_post_load_options]
+ * (
+ 1
+ if loader_opt is selectinload or operation_should_fail
+ else 4
+ )
+ ),
+ )
+
def test_override_parameters_scalar(self):
"""test that session.scalar() maintains the 'scalar-ness' of the
result when using re-execute events.
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
