Python collections 模块，Awaitable() 实例源码

我们从Python开源项目中，提取了以下22个代码示例，用于说明如何使用collections.Awaitable()。

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

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

项目：teleport 作者：eomsoft | 项目源码 | 文件源码

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

项目：TornadoWeb 作者：VxCoder | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

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

项目：teleport 作者：eomsoft | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：teleport 作者：eomsoft | 项目源码 | 文件源码

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

项目：TornadoWeb 作者：VxCoder | 项目源码 | 文件源码

def isawaitable(obj):
        return isinstance(obj, Awaitable)


###
#  allow patching the stdlib

项目：TornadoWeb 作者：VxCoder | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

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

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

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

项目：teleport 作者：eomsoft | 项目源码 | 文件源码

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

项目：TornadoWeb 作者：VxCoder | 项目源码 | 文件源码

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