Method orm_pre_session_exec

lib/sqlalchemy/orm/context.py:506–593

(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        is_pre_event,
    )

    @classmethod
    def orm_pre_session_exec(
        cls,
        session,
        statement,
        params,
        execution_options,
        bind_arguments,
        is_pre_event,
    ):
        # consume result-level load_options.  These may have been set up
        # in an ORMExecuteState hook
        (
            load_options,
            execution_options,
        ) = QueryContext.default_load_options.from_execution_options(
            "_sa_orm_load_options",
            {
                "populate_existing",
                "autoflush",
                "yield_per",
                "identity_token",
                "sa_top_level_orm_context",
            },
            execution_options,
            statement._execution_options,
        )

        # default execution options for ORM results:
        # 1. _result_disable_adapt_to_context=True
        #    this will disable the ResultSetMetadata._adapt_to_context()
        #    step which we don't need, as we have result processors cached
        #    against the original SELECT statement before caching.

        if "sa_top_level_orm_context" in execution_options:
            ctx = execution_options["sa_top_level_orm_context"]
            execution_options = ctx.query._execution_options.merge_with(
                ctx.execution_options, execution_options
            )

        if not execution_options:
            execution_options = _orm_load_exec_options
        else:
            execution_options = execution_options.union(_orm_load_exec_options)

        # would have been placed here by legacy Query only
        if load_options._yield_per:
            execution_options = execution_options.union(
                {"yield_per": load_options._yield_per}
            )

        if (
            getattr(statement._compile_options, "_current_path", None)
            and len(statement._compile_options._current_path) > 10
            and execution_options.get("compiled_cache", True) is not None
        ):
            execution_options: util.immutabledict[str, Any] = (
                execution_options.union(
                    {