"""Functional utilities."""
from __future__ import absolute_import, unicode_literals
from .abstract import Thenable
from .promises import promise
__all__ = [
'maybe_promise', 'ensure_promise',
'ppartial', 'preplace', 'ready_promise',
'starpromise', 'transform', 'wrap',
]
[docs]def maybe_promise(p):
"""Return None if p is unefined, otherwise make sure it's a promise."""
if p:
if not isinstance(p, Thenable):
return promise(p)
return p
[docs]def ensure_promise(p):
"""Ensure p is a promise.
If p is not a promise, a new promise is created with p' as callback.
"""
if p is None:
return promise()
return maybe_promise(p)
[docs]def ppartial(p, *args, **kwargs):
"""Create/modify promise with partial arguments."""
p = ensure_promise(p)
if args:
p.args = args + p.args
if kwargs:
p.kwargs.update(kwargs)
return p
[docs]def preplace(p, *args, **kwargs):
"""Replace promise arguments.
This will force the promise to disregard any arguments
the promise is fulfilled with, and to be called with the
provided arguments instead.
"""
def _replacer(*_, **__):
return p(*args, **kwargs)
return promise(_replacer)
[docs]def ready_promise(callback=None, *args):
"""Create promise that is already fulfilled."""
p = ensure_promise(callback)
p(*args)
return p
[docs]def starpromise(fun, *args, **kwargs):
"""Create promise, using star arguments."""
return promise(fun, args, kwargs)
def _transback(filter_, callback, args, kwargs, ret):
try:
ret = filter_(*args + (ret,), **kwargs)
except Exception:
callback.throw()
else:
return callback(ret)
[docs]def wrap(p):
"""Wrap promise.
This wraps the promise such that if the promise is called with a promise as
argument, we attach ourselves to that promise instead.
"""
def on_call(*args, **kwargs):
if len(args) == 1 and isinstance(args[0], promise):
return args[0].then(p)
else:
return p(*args, **kwargs)
return on_call