diff --git a/stdlib/asyncio/tasks.pyi b/stdlib/asyncio/tasks.pyi index 4613bca70c1a..3ff3e6c6d985 100644 --- a/stdlib/asyncio/tasks.pyi +++ b/stdlib/asyncio/tasks.pyi @@ -3,6 +3,7 @@ import sys from collections.abc import Awaitable, Coroutine, Generator, Iterable, Iterator from types import FrameType from typing import Any, Literal, Protocol, TextIO, TypeVar, overload + from typing_extensions import TypeAlias from . import _CoroutineLike @@ -73,7 +74,9 @@ _FT = TypeVar("_FT", bound=Future[Any]) if sys.version_info >= (3, 12): _FutureLike: TypeAlias = Future[_T] | Awaitable[_T] else: - _FutureLike: TypeAlias = Future[_T] | Generator[Any, None, _T] | Awaitable[_T] + _FutureLike: TypeAlias = ( + Future[_T] | Generator[Any, None, _T] | Awaitable[_T] + ) _TaskYieldType: TypeAlias = Future[object] | None FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED @@ -81,17 +84,24 @@ FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION ALL_COMPLETED = concurrent.futures.ALL_COMPLETED if sys.version_info >= (3, 10): - def as_completed(fs: Iterable[_FutureLike[_T]], *, timeout: float | None = None) -> Iterator[Future[_T]]: ... + def as_completed( + fs: Iterable[_FutureLike[_T]], *, timeout: float | None = None + ) -> Iterator[Future[_T]]: ... else: def as_completed( - fs: Iterable[_FutureLike[_T]], *, loop: AbstractEventLoop | None = None, timeout: float | None = None + fs: Iterable[_FutureLike[_T]], + *, + loop: AbstractEventLoop | None = None, + timeout: float | None = None, ) -> Iterator[Future[_T]]: ... @overload def ensure_future(coro_or_future: _FT, *, loop: AbstractEventLoop | None = None) -> _FT: ... # type: ignore[overload-overlap] @overload -def ensure_future(coro_or_future: Awaitable[_T], *, loop: AbstractEventLoop | None = None) -> Task[_T]: ... +def ensure_future( + coro_or_future: Awaitable[_T], *, loop: AbstractEventLoop | None = None +) -> Task[_T]: ... # `gather()` actually returns a list with length equal to the number # of tasks passed; however, Tuple is used similar to the annotation for @@ -104,7 +114,11 @@ if sys.version_info >= (3, 10): def gather(coro_or_future1: _FutureLike[_T1], /, *, return_exceptions: Literal[False] = False) -> Future[tuple[_T1]]: ... # type: ignore[overload-overlap] @overload def gather( # type: ignore[overload-overlap] - coro_or_future1: _FutureLike[_T1], coro_or_future2: _FutureLike[_T2], /, *, return_exceptions: Literal[False] = False + coro_or_future1: _FutureLike[_T1], + coro_or_future2: _FutureLike[_T2], + /, + *, + return_exceptions: Literal[False] = False, ) -> Future[tuple[_T1, _T2]]: ... @overload def gather( # type: ignore[overload-overlap] @@ -154,7 +168,11 @@ if sys.version_info >= (3, 10): def gather(coro_or_future1: _FutureLike[_T1], /, *, return_exceptions: bool) -> Future[tuple[_T1 | BaseException]]: ... # type: ignore[overload-overlap] @overload def gather( # type: ignore[overload-overlap] - coro_or_future1: _FutureLike[_T1], coro_or_future2: _FutureLike[_T2], /, *, return_exceptions: bool + coro_or_future1: _FutureLike[_T1], + coro_or_future2: _FutureLike[_T2], + /, + *, + return_exceptions: bool, ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException]]: ... @overload def gather( # type: ignore[overload-overlap] @@ -164,7 +182,9 @@ if sys.version_info >= (3, 10): /, *, return_exceptions: bool, - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ... + ) -> Future[ + tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException] + ]: ... @overload def gather( # type: ignore[overload-overlap] coro_or_future1: _FutureLike[_T1], @@ -174,7 +194,14 @@ if sys.version_info >= (3, 10): /, *, return_exceptions: bool, - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ... + ) -> Future[ + tuple[ + _T1 | BaseException, + _T2 | BaseException, + _T3 | BaseException, + _T4 | BaseException, + ] + ]: ... @overload def gather( # type: ignore[overload-overlap] coro_or_future1: _FutureLike[_T1], @@ -186,7 +213,13 @@ if sys.version_info >= (3, 10): *, return_exceptions: bool, ) -> Future[ - tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException, _T5 | BaseException] + tuple[ + _T1 | BaseException, + _T2 | BaseException, + _T3 | BaseException, + _T4 | BaseException, + _T5 | BaseException, + ] ]: ... @overload def gather( # type: ignore[overload-overlap] @@ -210,12 +243,18 @@ if sys.version_info >= (3, 10): ] ]: ... @overload - def gather(*coros_or_futures: _FutureLike[_T], return_exceptions: bool) -> Future[list[_T | BaseException]]: ... + def gather( + *coros_or_futures: _FutureLike[_T], return_exceptions: bool + ) -> Future[list[_T | BaseException]]: ... else: @overload def gather( # type: ignore[overload-overlap] - coro_or_future1: _FutureLike[_T1], /, *, loop: AbstractEventLoop | None = None, return_exceptions: Literal[False] = False + coro_or_future1: _FutureLike[_T1], + /, + *, + loop: AbstractEventLoop | None = None, + return_exceptions: Literal[False] = False, ) -> Future[tuple[_T1]]: ... @overload def gather( # type: ignore[overload-overlap] @@ -274,11 +313,17 @@ else: ) -> Future[tuple[_T1, _T2, _T3, _T4, _T5, _T6]]: ... @overload def gather( # type: ignore[overload-overlap] - *coros_or_futures: _FutureLike[_T], loop: AbstractEventLoop | None = None, return_exceptions: Literal[False] = False + *coros_or_futures: _FutureLike[_T], + loop: AbstractEventLoop | None = None, + return_exceptions: Literal[False] = False, ) -> Future[list[_T]]: ... @overload def gather( # type: ignore[overload-overlap] - coro_or_future1: _FutureLike[_T1], /, *, loop: AbstractEventLoop | None = None, return_exceptions: bool + coro_or_future1: _FutureLike[_T1], + /, + *, + loop: AbstractEventLoop | None = None, + return_exceptions: bool, ) -> Future[tuple[_T1 | BaseException]]: ... @overload def gather( # type: ignore[overload-overlap] @@ -298,7 +343,9 @@ else: *, loop: AbstractEventLoop | None = None, return_exceptions: bool, - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ... + ) -> Future[ + tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException] + ]: ... @overload def gather( # type: ignore[overload-overlap] coro_or_future1: _FutureLike[_T1], @@ -309,7 +356,14 @@ else: *, loop: AbstractEventLoop | None = None, return_exceptions: bool, - ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ... + ) -> Future[ + tuple[ + _T1 | BaseException, + _T2 | BaseException, + _T3 | BaseException, + _T4 | BaseException, + ] + ]: ... @overload def gather( # type: ignore[overload-overlap] coro_or_future1: _FutureLike[_T1], @@ -334,10 +388,14 @@ else: ]: ... @overload def gather( - *coros_or_futures: _FutureLike[_T], loop: AbstractEventLoop | None = None, return_exceptions: bool + *coros_or_futures: _FutureLike[_T], + loop: AbstractEventLoop | None = None, + return_exceptions: bool, ) -> Future[list[_T | BaseException]]: ... -def run_coroutine_threadsafe(coro: _FutureLike[_T], loop: AbstractEventLoop) -> concurrent.futures.Future[_T]: ... +def run_coroutine_threadsafe( + coro: _FutureLike[_T], loop: AbstractEventLoop +) -> concurrent.futures.Future[_T]: ... if sys.version_info >= (3, 10): def shield(arg: _FutureLike[_T]) -> Future[_T]: ... @@ -348,31 +406,54 @@ if sys.version_info >= (3, 10): async def wait_for(fut: _FutureLike[_T], timeout: float | None) -> _T: ... else: - def shield(arg: _FutureLike[_T], *, loop: AbstractEventLoop | None = None) -> Future[_T]: ... - @overload - async def sleep(delay: float, *, loop: AbstractEventLoop | None = None) -> None: ... - @overload - async def sleep(delay: float, result: _T, *, loop: AbstractEventLoop | None = None) -> _T: ... - async def wait_for(fut: _FutureLike[_T], timeout: float | None, *, loop: AbstractEventLoop | None = None) -> _T: ... + def shield( + arg: _FutureLike[_T], *, loop: AbstractEventLoop | None = None + ) -> Future[_T]: ... + @overload + async def sleep( + delay: float, *, loop: AbstractEventLoop | None = None + ) -> None: ... + @overload + async def sleep( + delay: float, result: _T, *, loop: AbstractEventLoop | None = None + ) -> _T: ... + async def wait_for( + fut: _FutureLike[_T], + timeout: float | None, + *, + loop: AbstractEventLoop | None = None, + ) -> _T: ... if sys.version_info >= (3, 11): @overload async def wait( - fs: Iterable[_FT], *, timeout: float | None = None, return_when: str = "ALL_COMPLETED" + fs: Iterable[_FT], + *, + timeout: float | None = None, + return_when: str = "ALL_COMPLETED", ) -> tuple[set[_FT], set[_FT]]: ... @overload async def wait( - fs: Iterable[Task[_T]], *, timeout: float | None = None, return_when: str = "ALL_COMPLETED" + fs: Iterable[Task[_T]], + *, + timeout: float | None = None, + return_when: str = "ALL_COMPLETED", ) -> tuple[set[Task[_T]], set[Task[_T]]]: ... elif sys.version_info >= (3, 10): @overload async def wait( # type: ignore[overload-overlap] - fs: Iterable[_FT], *, timeout: float | None = None, return_when: str = "ALL_COMPLETED" + fs: Iterable[_FT], + *, + timeout: float | None = None, + return_when: str = "ALL_COMPLETED", ) -> tuple[set[_FT], set[_FT]]: ... @overload async def wait( - fs: Iterable[Awaitable[_T]], *, timeout: float | None = None, return_when: str = "ALL_COMPLETED" + fs: Iterable[Awaitable[_T]], + *, + timeout: float | None = None, + return_when: str = "ALL_COMPLETED", ) -> tuple[set[Task[_T]], set[Task[_T]]]: ... else: @@ -396,9 +477,13 @@ else: if sys.version_info >= (3, 12): _TaskCompatibleCoro: TypeAlias = Coroutine[Any, Any, _T_co] elif sys.version_info >= (3, 9): - _TaskCompatibleCoro: TypeAlias = Generator[_TaskYieldType, None, _T_co] | Coroutine[Any, Any, _T_co] + _TaskCompatibleCoro: TypeAlias = ( + Generator[_TaskYieldType, None, _T_co] | Coroutine[Any, Any, _T_co] + ) else: - _TaskCompatibleCoro: TypeAlias = Generator[_TaskYieldType, None, _T_co] | Awaitable[_T_co] + _TaskCompatibleCoro: TypeAlias = ( + Generator[_TaskYieldType, None, _T_co] | Awaitable[_T_co] + ) # mypy and pyright complain that a subclass of an invariant class shouldn't be covariant. # While this is true in general, here it's sort-of okay to have a covariant subclass, @@ -426,35 +511,52 @@ class Task(Future[_T_co]): # type: ignore[type-var] # pyright: ignore[reportIn ) -> None: ... else: def __init__( - self, coro: _TaskCompatibleCoro[_T_co], *, loop: AbstractEventLoop = ..., name: str | None = ... + self, + coro: _TaskCompatibleCoro[_T_co], + *, + loop: AbstractEventLoop = ..., + name: str | None = ..., ) -> None: ... - def get_coro(self) -> _TaskCompatibleCoro[_T_co]: ... + def get_coro(self) -> _TaskCompatibleCoro[_T_co] | None: ... def get_name(self) -> str: ... def set_name(self, value: object, /) -> None: ... if sys.version_info >= (3, 12): def get_context(self) -> Context: ... def get_stack(self, *, limit: int | None = None) -> list[FrameType]: ... - def print_stack(self, *, limit: int | None = None, file: TextIO | None = None) -> None: ... + def print_stack( + self, *, limit: int | None = None, file: TextIO | None = None + ) -> None: ... if sys.version_info >= (3, 11): def cancelling(self) -> int: ... def uncancel(self) -> int: ... if sys.version_info < (3, 9): @classmethod - def current_task(cls, loop: AbstractEventLoop | None = None) -> Task[Any] | None: ... + def current_task( + cls, loop: AbstractEventLoop | None = None + ) -> Task[Any] | None: ... @classmethod - def all_tasks(cls, loop: AbstractEventLoop | None = None) -> set[Task[Any]]: ... + def all_tasks( + cls, loop: AbstractEventLoop | None = None + ) -> set[Task[Any]]: ... if sys.version_info >= (3, 9): def __class_getitem__(cls, item: Any, /) -> GenericAlias: ... def all_tasks(loop: AbstractEventLoop | None = None) -> set[Task[Any]]: ... if sys.version_info >= (3, 11): - def create_task(coro: _CoroutineLike[_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]: ... else: - def create_task(coro: _CoroutineLike[_T], *, name: str | None = None) -> Task[_T]: ... + def create_task( + coro: _CoroutineLike[_T], *, name: str | None = None + ) -> Task[_T]: ... def current_task(loop: AbstractEventLoop | None = None) -> Task[Any] | None: ... def _enter_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...