我们从Python开源项目中，提取了以下27个代码示例，用于说明如何使用collections.abc.Coroutine()。
def patch(patch_inspect=True): """ Main entry point for patching the ``collections.abc`` and ``inspect`` standard library modules. """ PATCHED['collections.abc.Generator'] = _collections_abc.Generator = Generator PATCHED['collections.abc.Coroutine'] = _collections_abc.Coroutine = Coroutine PATCHED['collections.abc.Awaitable'] = _collections_abc.Awaitable = Awaitable if patch_inspect: import inspect PATCHED['inspect.isawaitable'] = inspect.isawaitable = isawaitable
def register(self, event: Event, handler: Coroutine): self.events[event].append(handler)
def get_callbacks(self, event: Event) -> List[Coroutine]: return self.events[event]
def register_event_callback(self, event: Event, callback: Coroutine): """ Register a new event callback. """ self._events.register(event, callback)
def async_generator(func: Callable[..., Coroutine]) -> Callable[..., AsyncIterator]: warn('This function has been deprecated in favor of async_generator.async_generator', DeprecationWarning) return async_generator_(func)
def coroutine(func): """Convert regular generator function to a coroutine.""" if not callable(func): raise TypeError('types.coroutine() expects a callable') if (func.__class__ is FunctionType and getattr(func, '__code__', None).__class__ is CodeType): co_flags = func.__code__.co_flags # Check if 'func' is a coroutine function. # (0x180 == CO_COROUTINE | CO_ITERABLE_COROUTINE) if co_flags & 0x180: return func # Check if 'func' is a generator function. # (0x20 == CO_GENERATOR) if co_flags & 0x20: # TODO: Implement this in C. co = func.__code__ func.__code__ = CodeType( co.co_argcount, co.co_kwonlyargcount, co.co_nlocals, co.co_stacksize, co.co_flags | 0x100, # 0x100 == CO_ITERABLE_COROUTINE co.co_code, co.co_consts, co.co_names, co.co_varnames, co.co_filename, co.co_name, co.co_firstlineno, co.co_lnotab, co.co_freevars, co.co_cellvars) return func # The following code is primarily to support functions that # return generator-like objects (for instance generators # compiled with Cython). @_functools.wraps(func) def wrapped(*args, **kwargs): coro = func(*args, **kwargs) if (coro.__class__ is CoroutineType or coro.__class__ is GeneratorType and coro.gi_code.co_flags & 0x100): # 'coro' is a native coroutine object or an iterable coroutine return coro if (isinstance(coro, _collections_abc.Generator) and not isinstance(coro, _collections_abc.Coroutine)): # 'coro' is either a pure Python generator iterator, or it # implements collections.abc.Generator (and does not implement # collections.abc.Coroutine). return _GeneratorWrapper(coro) # 'coro' is either an instance of collections.abc.Coroutine or # some other object -- pass it through. return coro return wrapped
def mk_coroutine(): from abc import abstractmethod class Coroutine(Awaitable): __slots__ = () @abstractmethod def send(self, value): """Send a value into the coroutine. Return next yielded value or raise StopIteration. """ raise StopIteration @abstractmethod def throw(self, typ, val=None, tb=None): """Raise an exception in the coroutine. Return next yielded value or raise StopIteration. """ if val is None: if tb is None: raise typ val = typ() if tb is not None: val = val.with_traceback(tb) raise val def close(self): """Raise GeneratorExit inside coroutine. """ try: self.throw(GeneratorExit) except (GeneratorExit, StopIteration): pass else: raise RuntimeError('coroutine ignored GeneratorExit') @classmethod def __subclasshook__(cls, C): if cls is Coroutine: mro = C.__mro__ for method in ('__await__', 'send', 'throw', 'close'): for base in mro: if method in base.__dict__: break else: return NotImplemented return True return NotImplemented return Coroutine ### # make all ABCs available in this module
