Source code for vaex.delayed

__author__ = 'maartenbreddels'
import aplus
import os

"""
Mini library to make working with promises/futures a bit more Python like.

Example:

@delayed
def f(grid):
    return grid**2

f(ds.count(delay=True))

See tests/delayed_test.py for more examples
"""


def promisify(value):
    # TODO, support futures etc..
    if isinstance(value, aplus.Promise):
        return value
    if isinstance(value, (list, tuple)):
        return aplus.listPromise(*list([promisify(k) for k in value]))
    if isinstance(value, dict):
        return aplus.dictPromise({k: promisify(v) for k, v in value.items()})
    else:
        return aplus.Promise.fulfilled(value)


[docs]def delayed(f): '''Decorator to transparantly accept delayed computation. Example: >>> delayed_sum = ds.sum(ds.E, binby=ds.x, limits=limits, >>> shape=4, delay=True) >>> @vaex.delayed >>> def total_sum(sums): >>> return sums.sum() >>> sum_of_sums = total_sum(delayed_sum) >>> ds.execute() >>> sum_of_sums.get() See the tutorial for a more complete example https://docs.vaex.io/en/latest/tutorial.html#Parallel-computations ''' def wrapped(*args, **kwargs): # print "calling", f, "with", kwargs # key_values = kwargs.items() key_promise = list([(key, promisify(value)) for key, value in kwargs.items()]) # key_promise = [(key, promisify(value)) for key, value in key_values] arg_promises = list([promisify(value) for value in args]) kwarg_promises = list([promise for key, promise in key_promise]) promises = arg_promises + kwarg_promises for promise in promises: def echo_error(exc, promise=promise): print("error with ", promise, "exception is", exc) # raise exc def echo(value, promise=promise): print("done with ", repr(promise), "value is", value) # promise.then(echo, echo_error) # print promises allarguments = aplus.listPromise(*promises) def call(_): kwargs_real = {key: promise.get() for key, promise in key_promise} args_real = list([promise.get() for promise in arg_promises]) return f(*args_real, **kwargs_real) def error(exc): if os.environ.get('VAEX_DEBUG', False): print("Error from delayed", exc) # import vaex # vaex.utils.print_stack_trace() raise exc return allarguments.then(call, error) return wrapped
@delayed def delayed_args(*args): return args @delayed def delayed_kwargs(**kwargs): return kwargs @delayed def delayed_list(l): return delayed_args(*l) @delayed def delayed_dict(d): return delayed_kwargs(**d) @delayed def delayed_apply(f, args, kwargs): @delayed def internal(f, args, kwargs): return f(*args, **kwargs) return internal(f, args, kwargs)