[asyncio] Make create_task work with coroutines only in 3.12 (#10177)

Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com>

stdlib/asyncio/__init__.pyi

@@ -1,4 +1,7 @@
 import sys
+from collections.abc import Coroutine, Generator
+from typing import Any, TypeVar
+from typing_extensions import TypeAlias
 
 # As at runtime, this depends on all submodules defining __all__ accurately.
 from .base_events import *
@@ -28,3 +31,11 @@ if sys.platform == "win32":
     from .windows_events import *
 else:
     from .unix_events import *
+
+_T = TypeVar("_T")
+
+# Aliases imported by multiple submodules in typeshed
+if sys.version_info >= (3, 12):
+    _CoroutineLike: TypeAlias = Coroutine[Any, Any, _T]
+else:
+    _CoroutineLike: TypeAlias = Generator[Any, None, _T] | Coroutine[Any, Any, _T]

stdlib/asyncio/base_events.pyi

@@ -1,12 +1,13 @@
 import ssl
 import sys
 from _typeshed import FileDescriptorLike, ReadableBuffer, WriteableBuffer
+from asyncio import _CoroutineLike
 from asyncio.events import AbstractEventLoop, AbstractServer, Handle, TimerHandle, _TaskFactory
 from asyncio.futures import Future
 from asyncio.protocols import BaseProtocol
 from asyncio.tasks import Task
 from asyncio.transports import BaseTransport, DatagramTransport, ReadTransport, SubprocessTransport, Transport, WriteTransport
-from collections.abc import Awaitable, Callable, Coroutine, Generator, Iterable, Sequence
+from collections.abc import Awaitable, Callable, Generator, Iterable, Sequence
 from contextvars import Context
 from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
 from typing import IO, Any, TypeVar, overload
@@ -86,13 +87,11 @@ class BaseEventLoop(AbstractEventLoop):
     def create_future(self) -> Future[Any]: ...
     # Tasks methods
     if sys.version_info >= (3, 11):
-        def create_task(
-            self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T], *, name: object = None, context: Context | None = None
-        ) -> Task[_T]: ...
+        def create_task(self, coro: _CoroutineLike[_T], *, name: object = None, context: Context | None = None) -> Task[_T]: ...
     elif sys.version_info >= (3, 8):
-        def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T], *, name: object = None) -> Task[_T]: ...
+        def create_task(self, coro: _CoroutineLike[_T], *, name: object = None) -> Task[_T]: ...
     else:
-        def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T]) -> Task[_T]: ...
+        def create_task(self, coro: _CoroutineLike[_T]) -> Task[_T]: ...
 
     def set_task_factory(self, factory: _TaskFactory | None) -> None: ...
     def get_task_factory(self) -> _TaskFactory | None: ...

stdlib/asyncio/events.pyi

@@ -8,6 +8,7 @@ from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
 from typing import IO, Any, Protocol, TypeVar, overload
 from typing_extensions import Literal, Self, TypeAlias
 
+from . import _CoroutineLike
 from .base_events import Server
 from .futures import Future
 from .protocols import BaseProtocol
@@ -158,20 +159,14 @@ class AbstractEventLoop:
     if sys.version_info >= (3, 11):
         @abstractmethod
         def create_task(
-            self,
-            coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T],
-            *,
-            name: str | None = None,
-            context: Context | None = None,
+            self, coro: _CoroutineLike[_T], *, name: str | None = None, context: Context | None = None
         ) -> Task[_T]: ...
     elif sys.version_info >= (3, 8):
         @abstractmethod
-        def create_task(
-            self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T], *, name: str | None = None
-        ) -> Task[_T]: ...
+        def create_task(self, coro: _CoroutineLike[_T], *, name: str | None = None) -> Task[_T]: ...
     else:
         @abstractmethod
-        def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T]) -> Task[_T]: ...
+        def create_task(self, coro: _CoroutineLike[_T]) -> Task[_T]: ...
 
     @abstractmethod
     def set_task_factory(self, factory: _TaskFactory | None) -> None: ...

stdlib/asyncio/tasks.pyi

@@ -1,10 +1,11 @@
 import concurrent.futures
 import sys
-from collections.abc import Awaitable, Coroutine, Generator, Iterable, Iterator
+from collections.abc import Awaitable, Generator, Iterable, Iterator
 from types import FrameType
 from typing import Any, Generic, TextIO, TypeVar, overload
 from typing_extensions import Literal, TypeAlias
 
+from . import _CoroutineLike
 from .events import AbstractEventLoop
 from .futures import Future
 
@@ -312,15 +313,13 @@ class Task(Future[_T_co], Generic[_T_co]):  # type: ignore[type-var]  # pyright:
 def all_tasks(loop: AbstractEventLoop | None = None) -> set[Task[Any]]: ...
 
 if sys.version_info >= (3, 11):
-    def create_task(
-        coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = None, context: Context | None = None
-    ) -> Task[_T]: ...
+    def create_task(coro: _CoroutineLike[_T], *, name: str | None = None, context: Context | None = None) -> Task[_T]: ...
 
 elif sys.version_info >= (3, 8):
-    def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = None) -> Task[_T]: ...
+    def create_task(coro: _CoroutineLike[_T], *, name: str | None = None) -> Task[_T]: ...
 
 else:
-    def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T]) -> Task[_T]: ...
+    def create_task(coro: _CoroutineLike[_T]) -> Task[_T]: ...
 
 def current_task(loop: AbstractEventLoop | None = None) -> Task[Any] | None: ...
 def _enter_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...