diff --git a/stdlib/asyncio/tasks.pyi b/stdlib/asyncio/tasks.pyi index d7119b0400ba..25e4d57a25c7 100644 --- a/stdlib/asyncio/tasks.pyi +++ b/stdlib/asyncio/tasks.pyi @@ -3,7 +3,7 @@ import sys from collections.abc import Awaitable, Coroutine, Generator, Iterable, Iterator from types import FrameType from typing import Any, Generic, TextIO, TypeVar, overload -from typing_extensions import Literal, TypeAlias +from typing_extensions import Literal, TypeAlias, TypeVarTuple, Unpack from .events import AbstractEventLoop from .futures import Future @@ -58,6 +58,7 @@ _T3 = TypeVar("_T3") _T4 = TypeVar("_T4") _T5 = TypeVar("_T5") _FT = TypeVar("_FT", bound=Future[Any]) +_Ts = TypeVarTuple("_Ts") _FutureT: TypeAlias = Future[_T] | Generator[Any, None, _T] | Awaitable[_T] _TaskYieldType: TypeAlias = Future[object] | None @@ -81,191 +82,9 @@ def ensure_future(coro_or_future: Awaitable[_T], *, loop: AbstractEventLoop | No # Prior to Python 3.7 'async' was an alias for 'ensure_future'. # It became a keyword in 3.7. -# `gather()` actually returns a list with length equal to the number -# of tasks passed; however, Tuple is used similar to the annotation for -# zip() because typing does not support variadic type variables. See -# typing PR #1550 for discussion. -if sys.version_info >= (3, 10): - @overload - def gather(__coro_or_future1: _FutureT[_T1], *, return_exceptions: Literal[False] = ...) -> Future[tuple[_T1]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], __coro_or_future2: _FutureT[_T2], *, return_exceptions: Literal[False] = ... - ) -> Future[tuple[_T1, _T2]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - *, - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - *, - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3, _T4]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - __coro_or_future5: _FutureT[_T5], - *, - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3, _T4, _T5]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[Any], - __coro_or_future2: _FutureT[Any], - __coro_or_future3: _FutureT[Any], - __coro_or_future4: _FutureT[Any], - __coro_or_future5: _FutureT[Any], - __coro_or_future6: _FutureT[Any], - *coros_or_futures: _FutureT[Any], - return_exceptions: bool = ..., - ) -> Future[list[Any]]: ... - @overload - def gather(__coro_or_future1: _FutureT[_T1], *, return_exceptions: bool = ...) -> Future[tuple[_T1 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], __coro_or_future2: _FutureT[_T2], *, return_exceptions: bool = ... - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - *, - return_exceptions: bool = ..., - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - *, - return_exceptions: bool = ..., - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - __coro_or_future5: _FutureT[_T5], - *, - return_exceptions: bool = ..., - ) -> Future[ - tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException, _T5 | BaseException] - ]: ... - -else: - @overload - def gather( - __coro_or_future1: _FutureT[_T1], *, loop: AbstractEventLoop | None = ..., return_exceptions: Literal[False] = ... - ) -> Future[tuple[_T1]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3, _T4]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - __coro_or_future5: _FutureT[_T5], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: Literal[False] = ..., - ) -> Future[tuple[_T1, _T2, _T3, _T4, _T5]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[Any], - __coro_or_future2: _FutureT[Any], - __coro_or_future3: _FutureT[Any], - __coro_or_future4: _FutureT[Any], - __coro_or_future5: _FutureT[Any], - __coro_or_future6: _FutureT[Any], - *coros_or_futures: _FutureT[Any], - loop: AbstractEventLoop | None = ..., - return_exceptions: bool = ..., - ) -> Future[list[Any]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], *, loop: AbstractEventLoop | None = ..., return_exceptions: bool = ... - ) -> Future[tuple[_T1 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: bool = ..., - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: bool = ..., - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: bool = ..., - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ... - @overload - def gather( - __coro_or_future1: _FutureT[_T1], - __coro_or_future2: _FutureT[_T2], - __coro_or_future3: _FutureT[_T3], - __coro_or_future4: _FutureT[_T4], - __coro_or_future5: _FutureT[_T5], - *, - loop: AbstractEventLoop | None = ..., - return_exceptions: bool = ..., - ) -> Future[ - tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException, _T5 | BaseException] - ]: ... - +def gather( + *__coro_or_future: Unpack[_Ts[_FutureT[_T]]], return_exceptions: Literal[False] = ... +) -> Future[tuple[Unpack[_Ts[_T1]]]]: ... def run_coroutine_threadsafe(coro: _FutureT[_T], loop: AbstractEventLoop) -> concurrent.futures.Future[_T]: ... if sys.version_info >= (3, 10):