Threading and multiprocessing eye-candy.

Navigation

Verified details

These details have been verified by PyPI
Maintainers
Avatar for noxdafox from gravatar.com noxdafox

Unverified details

These details have not been verified by PyPI
Project links
Meta
  • License: GNU Library or Lesser General Public License (LGPL) (LGPL)
  • Author: Matteo Cafasso
  • Tags thread, process, pool, decorator
  • Requires: Python >=3.8
Classifiers

Project description

Pebble provides a neat API to manage threads and processes within an application.

Source:

https://github.com/noxdafox/pebble

Documentation:

https://pebble.readthedocs.io

Download:

https://pypi.org/project/Pebble/

Build Status Documentation Status PyPI - Downloads

Examples

Run a job in a separate thread and wait for its results.

from pebble import concurrent

@concurrent.thread
def function(foo, bar=0):
    return foo + bar

future = function(1, bar=2)

result = future.result()  # blocks until results are ready

Same code with AsyncIO support.

import asyncio

from pebble import asynchronous

@asynchronous.thread
def function(foo, bar=0):
    return foo + bar

async def asynchronous_function():
    result = await function(1, bar=2)  # blocks until results are ready
    print(result)

asyncio.run(asynchronous_function())

Run a function with a timeout of ten seconds and deal with errors.

from pebble import concurrent
from concurrent.futures import TimeoutError

@concurrent.process(timeout=10)
def function(foo, bar=0):
    return foo + bar

future = function(1, bar=2)

try:
    result = future.result()  # blocks until results are ready
except TimeoutError as error:
    print("Function took longer than %d seconds" % error.args[1])
except Exception as error:
    print("Function raised %s" % error)
    print(error.traceback)  # traceback of the function

Pools support workers restart, timeout for long running tasks and more.

from pebble import ProcessPool
from concurrent.futures import TimeoutError

TIMEOUT_SECONDS = 3

def function(foo, bar=0):
    return foo + bar

def task_done(future):
    try:
        result = future.result()  # blocks until results are ready
    except TimeoutError as error:
        print("Function took longer than %d seconds" % error.args[1])
    except Exception as error:
        print("Function raised %s" % error)
        print(error.traceback)  # traceback of the function

with ProcessPool(max_workers=5, max_tasks=10) as pool:
    for index in range(0, 10):
        future = pool.schedule(function, index, bar=1, timeout=TIMEOUT_SECONDS)
        future.add_done_callback(task_done)

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page