monkey_wrench.process._process module

The module providing functionalities for multiprocessing.

monkey_wrench.process._process._wrap_function_no_return(function: Callable[[T], R], argument: T, temporary_directory_path: Path) → None

Wrap the function by setting the global temporary directory.

The function is to be run in a separate spawned (and not forked) process!

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

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)])

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.

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

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).

run(function: Callable[[T], R], arguments: list[T] | set[T] | tuple[T, ...], temp_directory_path: Path | None = None) → None

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

If the function returns anything, it will be discarded!