我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用types.coroutine()。
def sleep(duration): """Return a `.Future` that resolves after the given number of seconds. When used with ``yield`` in a coroutine, this is a non-blocking analogue to `time.sleep` (which should not be used in coroutines because it is blocking):: yield gen.sleep(0.5) Note that calling this function on its own does nothing; you must wait on the `.Future` it returns (usually by yielding it). .. versionadded:: 4.1 """ f = Future() IOLoop.current().call_later(duration, lambda: f.set_result(None)) return f
def _argument_adapter(callback): """Returns a function that when invoked runs ``callback`` with one arg. If the function returned by this function is called with exactly one argument, that argument is passed to ``callback``. Otherwise the args tuple and kwargs dict are wrapped in an `Arguments` object. """ def wrapper(*args, **kwargs): if kwargs or len(args) > 1: callback(Arguments(args, kwargs)) elif args: callback(args[0]) else: callback(None) return wrapper # Convert Awaitables into Futures. It is unfortunately possible # to have infinite recursion here if those Awaitables assume that # we're using a different coroutine runner and yield objects # we don't understand. If that happens, the solution is to # register that runner's yieldable objects with convert_yielded.
def __init__(self, gen, result_future, first_yielded): self.gen = gen self.result_future = result_future self.future = _null_future self.yield_point = None self.pending_callbacks = None self.results = None self.running = False self.finished = False self.had_exception = False self.io_loop = IOLoop.current() # For efficiency, we do not create a stack context until we # reach a YieldPoint (stack contexts are required for the historical # semantics of YieldPoints, but not for Futures). When we have # done so, this field will be set and must be called at the end # of the coroutine. self.stack_context_deactivate = None if self.handle_yield(first_yielded): self.run()
def _value_from_stopiteration(e): try: # StopIteration has a value attribute beginning in py33. # So does our Return class. return e.value except AttributeError: pass try: # Cython backports coroutine functionality by putting the value in # e.args[0]. return e.args[0] except (AttributeError, IndexError): return None
def engine(func): """Callback-oriented decorator for asynchronous generators. This is an older interface; for new code that does not need to be compatible with versions of Tornado older than 3.0 the `coroutine` decorator is recommended instead. This decorator is similar to `coroutine`, except it does not return a `.Future` and the ``callback`` argument is not treated specially. In most cases, functions decorated with `engine` should take a ``callback`` argument and invoke it with their result when they are finished. One notable exception is the `~tornado.web.RequestHandler` :ref:`HTTP verb methods <verbs>`, which use ``self.finish()`` in place of a callback argument. """ func = _make_coroutine_wrapper(func, replace_callback=False) @functools.wraps(func) def wrapper(*args, **kwargs): future = func(*args, **kwargs) def final_callback(future): if future.result() is not None: raise ReturnValueIgnoredError( "@gen.engine functions cannot return values: %r" % (future.result(),)) # The engine interface doesn't give us any way to return # errors but to raise them into the stack context. # Save the stack context here to use when the Future has resolved. future.add_done_callback(stack_context.wrap(final_callback)) return wrapper
def coroutine(func, replace_callback=True): """Decorator for asynchronous generators. Any generator that yields objects from this module must be wrapped in either this decorator or `engine`. Coroutines may "return" by raising the special exception `Return(value) <Return>`. In Python 3.3+, it is also possible for the function to simply use the ``return value`` statement (prior to Python 3.3 generators were not allowed to also return values). In all versions of Python a coroutine that simply wishes to exit early may use the ``return`` statement without a value. Functions with this decorator return a `.Future`. Additionally, they may be called with a ``callback`` keyword argument, which will be invoked with the future's result when it resolves. If the coroutine fails, the callback will not be run and an exception will be raised into the surrounding `.StackContext`. The ``callback`` argument is not visible inside the decorated function; it is handled by the decorator itself. From the caller's perspective, ``@gen.coroutine`` is similar to the combination of ``@return_future`` and ``@gen.engine``. .. warning:: When exceptions occur inside a coroutine, the exception information will be stored in the `.Future` object. You must examine the result of the `.Future` object, or the exception may go unnoticed by your code. This means yielding the function if called from another coroutine, using something like `.IOLoop.run_sync` for top-level calls, or passing the `.Future` to `.IOLoop.add_future`. """ return _make_coroutine_wrapper(func, replace_callback=True)
def on_join(self, channel: str=None) -> Callable[[Callable], Callable]: """ Register a handler that's called after a channel is joined. The handler is called with the `Channel` as argument, must be a coroutine and is run non-blocking. :param channel: channel to look out for or `None` for all channels """ def decorator(fn: Callable[[self.IrcMessage], None]): jh = self.JoinHandler(channel, fn) self._on_join_handlers.append(jh) self._log.debug("Added join handler {}".format(jh)) return fn return decorator
def on_command(self, *args: Sequence[str], rest: str=None) -> Callable[[Callable], Callable]: """ Register a handler that's called when (the beginning of) a `IrcMessage` matches. The handler is called with the `IrcMessage` as argument, must be a coroutine and is run blocking, i.e. you cannot use `await_command` in it! :param args: commands args that must match (the actual command is the first arg) :param rest: match the rest (after the " :") of the `IrcMessage` """ def decorator(fn: Callable[[self.IrcMessage], None]): ch = self.CommandHandler(args, rest, fn) self._on_command_handlers.append(ch) self._log.debug("Added command handler {}".format(ch)) return fn return decorator
def _bg(self, coro: coroutine) -> asyncio.Task: """Run coro in background, log errors""" async def runner(): try: await coro except: self._log.exception("async: Coroutine raised exception") return asyncio.ensure_future(runner())
def asyncfunction(fn): # Set the coroutine flag fn = types.coroutine(fn) # Then wrap it in an 'async def', to enable the "coroutine was not # awaited" warning @wraps(fn) async def wrapper(*args, **kwargs): return await fn(*args, **kwargs) return wrapper # This class object is used as a singleton. # Not exported in the trio._core namespace, but imported directly by _run.
def sleep(seconds): """ Basic generator coroutine which another coroutine can await on to yield to the event loop for a specified time. """ print('sleep: Please wait {0} seconds before resuming.'.format(seconds)) yield seconds
def example(name): print('{}: Starting coroutine.'.format(name)) await sleep(0) print('{}: Resuming coroutine after first await'.format(name)) await sleep(5) print('{}: Resuming coroutine after 5 second sleep'.format(name))
def debug_wrapper(gen): # This function is called from 'sys.set_coroutine_wrapper'. # We only wrap here coroutines defined via 'async def' syntax. # Generator-based coroutines are wrapped in @coroutine # decorator. return CoroWrapper(gen, None)
def __init__(self, gen, func=None): assert inspect.isgenerator(gen) or inspect.iscoroutine(gen), gen self.gen = gen self.func = func # Used to unwrap @coroutine decorator self._source_traceback = traceback.extract_stack(sys._getframe(1)) self.__name__ = getattr(gen, '__name__', None) self.__qualname__ = getattr(gen, '__qualname__', None)
def __await__(self): cr_await = getattr(self.gen, 'cr_await', None) if cr_await is not None: raise RuntimeError( "Cannot await on coroutine {!r} while it's " "awaiting for {!r}".format(self.gen, cr_await)) return self
def iscoroutinefunction(func): """Return True if func is a decorated coroutine function.""" return (getattr(func, '_is_coroutine', None) is _is_coroutine or _inspect_iscoroutinefunction(func))
