monkey_wrench.process package

The package providing functionalities to run processes, e.g. multiprocessing.

class monkey_wrench.process.MultiProcess(*, number_of_processes: Annotated[int, Ge(ge=0)] = 1)[source]

Bases: Model

Pydantic model for multiprocessing.

Example

def power(x):            # Note that the function only accepts a single argument!
    print(x[0] ** x[1])  # We use indices to extract our desired arguments from the single input argument.

MultiProcess(number_of_processes=2).run(power, [(1, 3), (2, 5)])
run(function: Callable[[T], R], arguments: list[T] | set[T] | tuple[T, ...], temp_directory_path: Path | None = None) → None[source]

Similar to run_with_results(), but does not return anything.

If the function returns anything, it will be discarded!

run_with_results(function: Callable[[T], R], arguments: list[T] | set[T] | tuple[T, ...]) → list[R][source]

Call the provided function with different arguments using multiple processes.

Parameters:

  • function – The function to be called. It must only accept a single argument and must not be a lambda. In case you need a function with several input arguments, your function can accept single a tuple which packs all the arguments. You can then unpack, index, or iterate over the input, inside the function body. See the example.

  • arguments – An iterable (list/set/tuple) of arguments that will be passed to the function.

Returns:

A list of returned results from the function in the same order as the given arguments (if not a set).

number_of_processes: Annotated[int, Ge(ge=0)]

Number of process to use. Defaults to 1.

A value of 1 disables multiprocessing. This is useful for e.g. testing purposes.

Submodules