packages/vaex-core/vaex/delayed.py
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):
            print("error", exc)
            raise exc
        return allarguments.then(call, error)
    return wrapped
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):
            print("error", exc)
            raise exc
        return allarguments.then(call, error)
    return wrapped
def delayed ( f ) : 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 ) : print ( "error" , exc ) raise exc return allarguments . then ( call , error ) return wrapped
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
Decorator to transparantly accept delayed computation .