我们从Python开源项目中，提取了以下25个代码示例，用于说明如何使用collections.abc.Awaitable()。
def mk_awaitable(): from abc import abstractmethod, ABCMeta @abstractmethod def __await__(self): yield @classmethod def __subclasshook__(cls, C): if cls is Awaitable: for B in C.__mro__: if '__await__' in B.__dict__: if B.__dict__['__await__']: return True break return NotImplemented # calling metaclass directly as syntax differs in Py2/Py3 Awaitable = ABCMeta('Awaitable', (), { '__slots__': (), '__await__': __await__, '__subclasshook__': __subclasshook__, }) return Awaitable
def schedule_handler(cls: Any, obj: Any, context: Dict, func: Any, interval: Optional[Union[str, int]]=None, timestamp: Optional[str]=None, timezone: Optional[str]=None) -> Any: async def handler() -> None: values = inspect.getfullargspec(func) kwargs = {k: values.defaults[i] for i, k in enumerate(values.args[len(values.args) - len(values.defaults):])} if values.defaults else {} routine = func(*(obj,), **kwargs) try: if isinstance(routine, Awaitable): await routine except Exception as e: pass context['_schedule_scheduled_functions'] = context.get('_schedule_scheduled_functions', []) context['_schedule_scheduled_functions'].append((interval, timestamp, timezone, func, handler)) start_func = cls.start_scheduler(cls, obj, context) return (await start_func) if start_func else None
def isawaitable(obj): return isinstance(obj, Awaitable) ### # allow patching the stdlib
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 __anext__(self) -> Awaitable: if self._last_served_index < len(self.services): service = self._AsyncServiceFuture( Service(self.services[self._last_served_index]) ) self._last_served_index += 1 else: raise StopAsyncIteration() return service
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
def error_handler(cls: Any, obj: Any, context: Dict, func: Any, status_code: int) -> Any: default_content_type = context.get('options', {}).get('http', {}).get('content_type', 'text/plain') default_charset = context.get('options', {}).get('http', {}).get('charset', 'utf-8') if default_content_type is not None and ";" in default_content_type: # for backwards compability try: default_charset = str([v for v in default_content_type.split(';') if 'charset=' in v][0]).replace('charset=', '').strip() default_content_type = str([v for v in default_content_type.split(';')][0]).strip() except IndexError: pass async def handler(request: web.Request) -> web.Response: values = inspect.getfullargspec(func) kwargs = {k: values.defaults[i] for i, k in enumerate(values.args[len(values.args) - len(values.defaults):])} if values.defaults else {} routine = func(*(obj, request,), **kwargs) return_value = (await routine) if isinstance(routine, Awaitable) else routine # type: Union[str, bytes, Dict, List, Tuple, web.Response, Response] if isinstance(return_value, Response): return return_value.get_aiohttp_response(context, default_content_type=default_content_type, default_charset=default_charset) status = int(status_code) headers = None if isinstance(return_value, dict): body = return_value.get('body') _status = return_value.get('status') # type: Optional[SupportsInt] if _status and isinstance(_status, (int, str, bytes)): status = int(_status) if return_value.get('headers'): headers = CIMultiDict(return_value.get('headers')) elif isinstance(return_value, list) or isinstance(return_value, tuple): _status = return_value[0] if _status and isinstance(_status, (int, str, bytes)): status = int(_status) body = return_value[1] if len(return_value) > 2: headers = CIMultiDict(return_value[2]) elif isinstance(return_value, web.Response): return return_value else: if return_value is None: return_value = '' body = return_value return Response(body=body, status=status, headers=headers, content_type=default_content_type, charset=default_charset).get_aiohttp_response(context) context['_http_error_handler'] = context.get('_http_error_handler', {}) context['_http_error_handler'][int(status_code)] = handler start_func = cls.start_server(obj, context) return (await start_func) if start_func else None
