Source code for wolframclient.utils.asyncio
from __future__ import absolute_import, print_function, unicode_literals
import asyncio
import functools
from wolframclient.utils.functional import first, iterate
[docs]def run_in_loop(cor, loop=None):
@functools.wraps(cor)
def wrapped(*args, **kwargs):
return get_event_loop(loop).run_until_complete(cor(*args, **kwargs))
return wrapped
[docs]def run_all(args, **opts):
done = tuple(iterate(*args))
if done and len(done) > 1:
return asyncio.ensure_future(asyncio.wait(done), **opts)
elif done:
return asyncio.ensure_future(first(done), **opts)
return done
[docs]def get_event_loop(loop=None):
try:
return loop or asyncio.get_event_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
return loop
[docs]def silence(*exceptions):
def wrap(fn):
@functools.wraps(fn)
def wrapper(*args, **kwargs):
try:
return fn(*args, **kwargs)
except tuple(exceptions):
pass
return wrapper
return wrap
if hasattr(asyncio, "create_task"):
create_task = asyncio.create_task
else:
def create_task(coro):
""" ensure_future using get_event_loop, so that it behaves similarly to
create_task, and gets the same signature.
"""
return asyncio.ensure_future(coro, loop=asyncio.get_event_loop())