Repository: dask/dask
File: dask/dataframe/io/io.py (Python)

Code before:
    if verify_meta:
        for (i, df) in enumerate(dfs):
            dsk[(name, i)] = (check_meta, df.key, meta, "from_delayed")
    else:
        for (i, df) in enumerate(dfs):
            dsk[(name, i)] = df.key

    if divisions is None or divisions == "sorted":
        divs = [None] * (len(dfs) + 1)
    else:
        divs = tuple(divisions)
        if len(divs) != len(dfs) + 1:
            raise ValueError("divisions should be a tuple of len(dfs) + 1")

    df = new_dd_object(dsk, name, meta, divs)

    if divisions == "sorted":
        from ..shuffle import compute_and_set_divisions

        df = compute_and_set_divisions(df)

    return df


@insert_meta_param_description
def from_callable(
    func,
    npartitions,
    *args,
    meta=None,
    divisions=None,
    prefix="from-function",
    io_prefix="blockwise-io",
):
    """Create Dask DataFrame from a data-creation function.

    A function (specified by the `func` argument) will be used to generate the
    partitions of a new DataFrame collection.

    Parameters
    ----------
    dfs : Callable
        A function that accepts a partition index (and any other args and/or kwargs),
        and returns a DataFrame object.
    npartitions : int
        The number of partitions in the resulting DataFrame collection.
    $META
    divisions : tuple, str, optional
        Partition boundaries along the index.
        For tuple, see https://docs.dask.org/en/latest/dataframe-design.html#partitions
        For string 'sorted' will compute the delayed values to find index

Review comment (bug):
Note that `from_callable` was added in this PR to facilitate testing/debugging.  It is not clear if this type of function belongs in the public API.

Code after:
            dsk[(name, i)] = (check_meta, df.key, meta, "from_delayed")
    else:
        for (i, df) in enumerate(dfs):
            dsk[(name, i)] = df.key

    if divisions is None or divisions == "sorted":
        divs = [None] * (len(dfs) + 1)
    else:
        divs = tuple(divisions)
        if len(divs) != len(dfs) + 1:
            raise ValueError("divisions should be a tuple of len(dfs) + 1")

    df = new_dd_object(dsk, name, meta, divs)

    if divisions == "sorted":
        from ..shuffle import compute_and_set_divisions

        df = compute_and_set_divisions(df)

    return df


def sorted_division_locations(seq, npartitions=None, chunksize=None):
    """Find division locations and values in sorted list

    Examples
    --------

    >>> L = ['A', 'B', 'C', 'D', 'E', 'F']
    >>> sorted_division_locations(L, chunksize=2)
    (['A', 'C', 'E', 'F'], [0, 2, 4, 6])

    >>> sorted_division_locations(L, chunksize=3)
    (['A', 'D', 'F'], [0, 3, 6])

    >>> L = ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C']
    >>> sorted_division_locations(L, chunksize=3)
    (['A', 'B', 'C'], [0, 4, 8])

    >>> sorted_division_locations(L, chunksize=2)
    (['A', 'B', 'C'], [0, 4, 8])

    >>> sorted_division_locations(['A'], chunksize=2)
    (['A', 'A'], [0, 1])
    """
    if (npartitions is None) == (chunksize is None):
        raise ValueError("Exactly one of npartitions and chunksize must be specified.")

    if npartitions:
        chunksize = ceil(len(seq) / npartitions)