Skip to content

Variadic Generics in asyncio.gather() #8085

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Harry-Lees wants to merge 4 commits into from
Closed

Variadic Generics in asyncio.gather() #8085

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c79730e
added support for arbitary length *args with variadic generic
Harry-Lees Jun 15, 2022
efb6421
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jun 15, 2022
817cddc
changed imports to typing_extensions to ensure compatibility with all…
Harry-Lees Jun 15, 2022
0e962d4
removed redundant type annotations for individual Python versions
Harry-Lees Jun 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 5 additions & 186 deletions stdlib/asyncio/tasks.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand All @@ -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):
Expand Down