PEP 0492 -- Coroutines with async and await syntax. Issue #24017.

Lib/_collections_abc.py

@@ -9,7 +9,8 @@ Unit tests are in test_collections.
 from abc import ABCMeta, abstractmethod
 import sys
 
-__all__ = ["Hashable", "Iterable", "Iterator", "Generator",
+__all__ = ["Awaitable", "Coroutine",
+           "Hashable", "Iterable", "Iterator", "Generator",
            "Sized", "Container", "Callable",
            "Set", "MutableSet",
            "Mapping", "MutableMapping",
@@ -74,6 +75,79 @@ class Hashable(metaclass=ABCMeta):
         return NotImplemented
 
 
+class _CoroutineMeta(ABCMeta):
+
+    def __instancecheck__(cls, instance):
+        # 0x80 = CO_COROUTINE
+        # 0x100 = CO_ITERABLE_COROUTINE
+        # We don't want to import 'inspect' module, as
+        # a dependency for 'collections.abc'.
+        CO_COROUTINES = 0x80 | 0x100
+
+        if (isinstance(instance, generator) and
+            instance.gi_code.co_flags & CO_COROUTINES):
+
+            return True
+
+        return super().__instancecheck__(instance)
+
+
+class Coroutine(metaclass=_CoroutineMeta):
+
+    __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")
+
+
+class Awaitable(metaclass=_CoroutineMeta):
+
+    __slots__ = ()
+
+    @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
+
+Awaitable.register(Coroutine)
+
+
 class Iterable(metaclass=ABCMeta):
 
     __slots__ = ()