pebble_process_pool_async_task_with_timeout.py

1from pebble import ProcessPool
2from concurrent.futures import TimeoutError
3
4def task(foo, bar=None):
5    return "done"
6
7with ProcessPool() as pool:
8    future = pool.schedule(task, args=["foo"], kwargs={"bar": "bar"}, timeout=10)
9
10    try:
11        result = future.result()  # blocks until results are ready
12    except TimeoutError as error:
13        print("Function took longer than %d seconds" % error.args[1])
14    except Exception as error:
15        print("Function raised %s" % error)
16        print(error.traceback)  # traceback of the function