Function collections_to_expr

dask/base.py:428–480

Convert many collections into a single dask expression. Typically, users should not be required to interact with this function. Parameters ---------- collections : Iterable An iterable of dask collections to be combined. optimize_graph : bool, optional If t

(
    collections: Iterable,
    optimize_graph: bool = True,
)

def collections_to_expr(
    collections: Iterable,
    optimize_graph: bool = True,
) -> Expr:
    """
    Convert many collections into a single dask expression.

    Typically, users should not be required to interact with this function.

    Parameters
    ----------
    collections : Iterable
        An iterable of dask collections to be combined.
    optimize_graph : bool, optional
        If this is True and collections are encountered which are backed by
        legacy HighLevelGraph objects, the returned Expression will run a low
        level task optimization during materialization.
    """
    is_iterable = False
    if isinstance(collections, (tuple, list, set)):
        is_iterable = True
    else:
        collections = [collections]
    if not collections:
        raise ValueError("No collections provided")
    from dask._expr import CompositeExpr, Expr, HLGExpr, _ExprSequence

    graphs = []
    for coll in collections:
        from dask.delayed import Delayed

        if isinstance(coll, Delayed):
            graphs.append(HLGExpr.from_collection(coll, optimize_graph=optimize_graph))
        else:
            expr = getattr(coll, "expr", None)
            if isinstance(expr, Expr):
                graphs.append(expr)
                continue

            dask_exprs = getattr(coll, "__dask_exprs__", None)
            if dask_exprs is not None:
                exprs = dask_exprs()
                if exprs is not None:
                    exprs = tuple(exprs)
                    if exprs:
                        graphs.append(CompositeExpr(coll, *exprs))
                        continue
            graphs.append(HLGExpr.from_collection(coll, optimize_graph=optimize_graph))

    if len(graphs) > 1 or is_iterable:
        return _ExprSequence(*graphs)
    else:
        return graphs[0]


def _rebuild_composite_collection(