Thread/Process runner helper. (pre async era)

concurrent_utils.py

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def task_runner(opts):
    if isinstance(opts, (list, tuple)):
        num_opts = len(opts)
        args = None
        kwargs = None
        if num_opts == 2:
            fn, args = opts
        elif num_opts == 3:
            fn, args, kwargs = opts
        return fn(*(args or []), **(kwargs or {}))
    if callable(opts):
        return opts()


def run_concurrent(funcs, run_on_process=False, max_workers=4):
    """ Run all funcs (fn, args, kwargs) and return result in order """
    ExeClass = ThreadPoolExecutor
    if run_on_process:
        ExeClass = ProcessPoolExecutor
    with ExeClass(max_workers=max_workers) as exe:
        results = exe.map(task_runner, funcs)
    return results