Function merge_frozen_result

lib/sqlalchemy/orm/loading.py:301–357

Merge a :class:`_engine.FrozenResult` back into a :class:`_orm.Session`, returning a new :class:`_engine.Result` object with :term:`persistent` objects. See the section :ref:`do_orm_execute_re_executing` for an example. .. seealso:: :ref:`do_orm_execute_re_executing`

(session, statement, frozen_result, load=True)

@util.preload_module("sqlalchemy.orm.context")
def merge_frozen_result(session, statement, frozen_result, load=True):
    """Merge a :class:`_engine.FrozenResult` back into a :class:`_orm.Session`,
    returning a new :class:`_engine.Result` object with :term:`persistent`
    objects.

    See the section :ref:`do_orm_execute_re_executing` for an example.

    .. seealso::

        :ref:`do_orm_execute_re_executing`

        :meth:`_engine.Result.freeze`

        :class:`_engine.FrozenResult`

    """
    querycontext = util.preloaded.orm_context

    if load:
        # flush current contents if we expect to load data
        session._autoflush()

    ctx = querycontext.ORMSelectCompileState._create_entities_collection(
        statement, legacy=False
    )

    autoflush = session.autoflush
    try:
        session.autoflush = False
        mapped_entities = [
            i
            for i, e in enumerate(ctx._entities)
            if isinstance(e, querycontext._MapperEntity)
        ]
        keys = [ent._label_name for ent in ctx._entities]

        keyed_tuple = result_tuple(
            keys, [ent._extra_entities for ent in ctx._entities]
        )

        result = []
        for newrow in frozen_result.rewrite_rows():
            for i in mapped_entities:
                if newrow[i] is not None:
                    newrow[i] = session._merge(
                        attributes.instance_state(newrow[i]),
                        attributes.instance_dict(newrow[i]),
                        load=load,
                        _recursive={},
                        _resolve_conflict_map={},
                    )

            result.append(keyed_tuple(newrow))

        return frozen_result.with_new_rows(result)
    finally:
        session.autoflush = autoflush