Function merge_result

lib/sqlalchemy/orm/loading.py:366–447

Merge a result into the given :class:`.Query` object's Session. See :meth:`_orm.Query.merge_result` for top-level documentation on this function.

(
    query: Query[Any],
    iterator: Union[FrozenResult, Iterable[Sequence[Any]], Iterable[object]],
    load: bool = True,
)

)
@util.preload_module("sqlalchemy.orm.context")
def merge_result(
    query: Query[Any],
    iterator: Union[FrozenResult, Iterable[Sequence[Any]], Iterable[object]],
    load: bool = True,
) -> Union[FrozenResult, Iterable[Any]]:
    """Merge a result into the given :class:`.Query` object's Session.

    See :meth:`_orm.Query.merge_result` for top-level documentation on this
    function.

    """

    querycontext = util.preloaded.orm_context

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

    # TODO: need test coverage and documentation for the FrozenResult
    # use case.
    if isinstance(iterator, FrozenResult):
        frozen_result = iterator
        iterator = iter(frozen_result.data)
    else:
        frozen_result = None

    ctx = querycontext.ORMSelectCompileState._create_entities_collection(
        query, legacy=True
    )

    autoflush = session.autoflush
    try:
        session.autoflush = False
        single_entity = not frozen_result and len(ctx._entities) == 1

        if single_entity:
            if isinstance(ctx._entities[0], querycontext._MapperEntity):
                result = [
                    session._merge(
                        attributes.instance_state(instance),
                        attributes.instance_dict(instance),
                        load=load,
                        _recursive={},
                        _resolve_conflict_map={},
                    )
                    for instance in iterator
                ]
            else:
                result = list(iterator)
        else:
            mapped_entities = [
                i
                for i, e in enumerate(ctx._entities)
                if isinstance(e, querycontext._MapperEntity)
            ]
            result = []
            keys = [ent._label_name for ent in ctx._entities]