I would like to make/have a scala like 'future' async API for python

asynchronous, python, scala

Solution

Python’s concurrent.futures indeed do a poor job in enabling future composition and reactive programming.

Have a look at this library, it implements Scala-like Futures and Promises. Implementation is small and well tested, and should be easy to port to Python 2.7 to suit your needs.

It wraps concurrent.futures *ThreadPoolExecutor* to return enhanced Future objects:

from rx.executors import ThreadPoolExecutor
from rx.futures import Future

with ThreadPoolExecutor(10) as tp:
    futures = [tp.submit(foo, i) for i in range(10)] # list of futures
    future_all = Future.all(futures) # combines into single future with list of results
    future_all_sum = future_all.map(sum) # maps result list to sum of elements
    print(future_all_sum.result(timeout=10))

Problem

I needed a few futures to wait on others in Python/Django and I had to hack my own "Promise" class using sources off the net and concurrent.futures (backport to 2.7). Here is my code (not pretty but works for exactly what I need), I got most of the idea from a blog post (and I cannot find the link to it now ... will update later): ``` class Promise(object): def __init__(self, func): self.func = func LOG.debug('Create Promise from callable %s %s' % (func, str(inspect.getargspec(func)))) def resolve_on_type(self, arg): if isinstance(arg, Future): return self.resolve_on_type(arg.result()) # happens when chaining two promises. elif isinstance(arg, types.GeneratorType): iterable = list() for a in arg: iterable.append(self.resolve_on_type(a)) return iterable else: return arg def resolve(self, *args, **kwargs): resolved_args = [] resolved_kwargs = {} #TODO need a more efficient way to wait on all results for i, arg in enumerate(args): resolved_args.append(self.resolve_on_type(arg)) for kw, arg in kwargs.items(): resolved_kwargs[kw] = self.resolve_on_type(arg) try: return self.func(*resolved_args, **resolved_kwargs) except: LOG.exception('<Promise> Error on task execution.') raise def __call__(self, *args, **kwargs): LOG.debug('Promise %s called' % self.func) return thread_pool.submit(self.resolve, *args, **kwargs) ``` Personally I would like something a bit more scala like. ``` def my_function(some, args, here): #do stuff that takes time and blocks or whatever return something future_1 = future(my_function, _some, _args, _here_please).map(mapping_function).recover(error_handling_function) future_1.result() list_of_futures = map(async_functions, some_args) future_of_list = sequence(list_of_futures) ``` I would appreciate any hints ... or is there something out there that works well? I must say `concurrent.futures` has simplified my task but I have no idea how get `map` work. I think I need experts (I have been less than a couple months in active python development before that I used it for maintenance scripts on files and dbs). I think if I get sequence and map I could get how to work the rest out for myself.

Original source