wove.helpers

wove.helpers contains small data-shaping utilities that make task dependencies easier to express. They are exported from wove for normal use.

Use helpers when a small list, dictionary, batch, or optional-value transformation belongs inside the weave but does not deserve a throwaway task function.

Common Helpers

• flatten: flatten nested iterables.

• fold: split an iterable into fixed-size groups.

• batch: split an iterable into a fixed number of groups.

• undict: turn a dictionary into key/value pairs.

• redict: rebuild a dictionary from key/value pairs.

• denone: remove None values.

• sync_to_async: wrap sync functions for async contexts.

API Details

wove.helpers.batch(a_list, count)

Converts a 1D list into count smaller lists of N length.

wove.helpers.denone(an_iterable)

Removes all None values from an iterable.

wove.helpers.flatten(list_of_lists)

Converts a 2D iterable into a 1D list.

wove.helpers.fold(a_list, size)

Converts a 1D list into N smaller lists of size length.

wove.helpers.redict(list_of_pairs)

Converts a list of key-value pairs back into a dictionary.

wove.helpers.sync_to_async(func)

Wraps a synchronous function to run in an executor.

By default, it uses asyncio’s default thread pool, but if called from within a weave context, it will use the dedicated executor for that context. This prevents deadlocks when a weave block is used inside another system that also manages the default executor (like Quart).

Parameters:

func (Callable[[...], R]) – The synchronous function to wrap.

Returns:

An awaitable coroutine function that executes the original function in a separate thread.

Return type:

Callable[[…], Coroutine[Any, Any, R]]

wove.helpers.undict(a_dict)

Converts a dictionary into a list of [key, value] pairs.