.config/coc/extensions/node_modules/coc-pyright/node_modules/pyright/dist/typeshed-fallback/stdlib/asyncio/tasks.pyi

   1 import concurrent.futures
   2 import sys
   3 from collections.abc import Awaitable, Generator, Iterable, Iterator
   4 from types import FrameType
   5 from typing import Any, Generic, Optional, TextIO, TypeVar, Union, overload
   6 from typing_extensions import Literal
   7
   8 from .events import AbstractEventLoop
   9 from .futures import Future
  10
  11 if sys.version_info >= (3, 9):
  12     from types import GenericAlias
  13
  14 _T = TypeVar("_T")
  15 _T1 = TypeVar("_T1")
  16 _T2 = TypeVar("_T2")
  17 _T3 = TypeVar("_T3")
  18 _T4 = TypeVar("_T4")
  19 _T5 = TypeVar("_T5")
  20 _FT = TypeVar("_FT", bound=Future[Any])
  21 _FutureT = Union[Future[_T], Generator[Any, None, _T], Awaitable[_T]]
  22 _TaskYieldType = Optional[Future[object]]
  23
  24 FIRST_EXCEPTION: str
  25 FIRST_COMPLETED: str
  26 ALL_COMPLETED: str
  27
  28 if sys.version_info >= (3, 10):
  29     def as_completed(fs: Iterable[_FutureT[_T]], *, timeout: float | None = ...) -> Iterator[Future[_T]]: ...
  30
  31 else:
  32     def as_completed(
  33         fs: Iterable[_FutureT[_T]], *, loop: AbstractEventLoop | None = ..., timeout: float | None = ...
  34     ) -> Iterator[Future[_T]]: ...
  35
  36 @overload
  37 def ensure_future(coro_or_future: _FT, *, loop: AbstractEventLoop | None = ...) -> _FT: ...  # type: ignore
  38 @overload
  39 def ensure_future(coro_or_future: Awaitable[_T], *, loop: AbstractEventLoop | None = ...) -> Task[_T]: ...
  40
  41 # Prior to Python 3.7 'async' was an alias for 'ensure_future'.
  42 # It became a keyword in 3.7.
  43
  44 # `gather()` actually returns a list with length equal to the number
  45 # of tasks passed; however, Tuple is used similar to the annotation for
  46 # zip() because typing does not support variadic type variables.  See
  47 # typing PR #1550 for discussion.
  48 if sys.version_info >= (3, 10):
  49     @overload
  50     def gather(coro_or_future1: _FutureT[_T1], *, return_exceptions: Literal[False] = ...) -> Future[tuple[_T1]]: ...
  51     @overload
  52     def gather(
  53         coro_or_future1: _FutureT[_T1], coro_or_future2: _FutureT[_T2], *, return_exceptions: Literal[False] = ...
  54     ) -> Future[tuple[_T1, _T2]]: ...
  55     @overload
  56     def gather(
  57         coro_or_future1: _FutureT[_T1],
  58         coro_or_future2: _FutureT[_T2],
  59         coro_or_future3: _FutureT[_T3],
  60         *,
  61         return_exceptions: Literal[False] = ...,
  62     ) -> Future[tuple[_T1, _T2, _T3]]: ...
  63     @overload
  64     def gather(
  65         coro_or_future1: _FutureT[_T1],
  66         coro_or_future2: _FutureT[_T2],
  67         coro_or_future3: _FutureT[_T3],
  68         coro_or_future4: _FutureT[_T4],
  69         *,
  70         return_exceptions: Literal[False] = ...,
  71     ) -> Future[tuple[_T1, _T2, _T3, _T4]]: ...
  72     @overload
  73     def gather(
  74         coro_or_future1: _FutureT[_T1],
  75         coro_or_future2: _FutureT[_T2],
  76         coro_or_future3: _FutureT[_T3],
  77         coro_or_future4: _FutureT[_T4],
  78         coro_or_future5: _FutureT[_T5],
  79         *,
  80         return_exceptions: Literal[False] = ...,
  81     ) -> Future[tuple[_T1, _T2, _T3, _T4, _T5]]: ...
  82     @overload
  83     def gather(
  84         coro_or_future1: _FutureT[Any],
  85         coro_or_future2: _FutureT[Any],
  86         coro_or_future3: _FutureT[Any],
  87         coro_or_future4: _FutureT[Any],
  88         coro_or_future5: _FutureT[Any],
  89         coro_or_future6: _FutureT[Any],
  90         *coros_or_futures: _FutureT[Any],
  91         return_exceptions: bool = ...,
  92     ) -> Future[list[Any]]: ...
  93     @overload
  94     def gather(coro_or_future1: _FutureT[_T1], *, return_exceptions: bool = ...) -> Future[tuple[_T1 | BaseException]]: ...
  95     @overload
  96     def gather(
  97         coro_or_future1: _FutureT[_T1], coro_or_future2: _FutureT[_T2], *, return_exceptions: bool = ...
  98     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException]]: ...
  99     @overload
 100     def gather(
 101         coro_or_future1: _FutureT[_T1],
 102         coro_or_future2: _FutureT[_T2],
 103         coro_or_future3: _FutureT[_T3],
 104         *,
 105         return_exceptions: bool = ...,
 106     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ...
 107     @overload
 108     def gather(
 109         coro_or_future1: _FutureT[_T1],
 110         coro_or_future2: _FutureT[_T2],
 111         coro_or_future3: _FutureT[_T3],
 112         coro_or_future4: _FutureT[_T4],
 113         *,
 114         return_exceptions: bool = ...,
 115     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ...
 116     @overload
 117     def gather(
 118         coro_or_future1: _FutureT[_T1],
 119         coro_or_future2: _FutureT[_T2],
 120         coro_or_future3: _FutureT[_T3],
 121         coro_or_future4: _FutureT[_T4],
 122         coro_or_future5: _FutureT[_T5],
 123         *,
 124         return_exceptions: bool = ...,
 125     ) -> Future[
 126         tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException, _T5 | BaseException]
 127     ]: ...
 128
 129 else:
 130     @overload
 131     def gather(
 132         coro_or_future1: _FutureT[_T1], *, loop: AbstractEventLoop | None = ..., return_exceptions: Literal[False] = ...
 133     ) -> Future[tuple[_T1]]: ...
 134     @overload
 135     def gather(
 136         coro_or_future1: _FutureT[_T1],
 137         coro_or_future2: _FutureT[_T2],
 138         *,
 139         loop: AbstractEventLoop | None = ...,
 140         return_exceptions: Literal[False] = ...,
 141     ) -> Future[tuple[_T1, _T2]]: ...
 142     @overload
 143     def gather(
 144         coro_or_future1: _FutureT[_T1],
 145         coro_or_future2: _FutureT[_T2],
 146         coro_or_future3: _FutureT[_T3],
 147         *,
 148         loop: AbstractEventLoop | None = ...,
 149         return_exceptions: Literal[False] = ...,
 150     ) -> Future[tuple[_T1, _T2, _T3]]: ...
 151     @overload
 152     def gather(
 153         coro_or_future1: _FutureT[_T1],
 154         coro_or_future2: _FutureT[_T2],
 155         coro_or_future3: _FutureT[_T3],
 156         coro_or_future4: _FutureT[_T4],
 157         *,
 158         loop: AbstractEventLoop | None = ...,
 159         return_exceptions: Literal[False] = ...,
 160     ) -> Future[tuple[_T1, _T2, _T3, _T4]]: ...
 161     @overload
 162     def gather(
 163         coro_or_future1: _FutureT[_T1],
 164         coro_or_future2: _FutureT[_T2],
 165         coro_or_future3: _FutureT[_T3],
 166         coro_or_future4: _FutureT[_T4],
 167         coro_or_future5: _FutureT[_T5],
 168         *,
 169         loop: AbstractEventLoop | None = ...,
 170         return_exceptions: Literal[False] = ...,
 171     ) -> Future[tuple[_T1, _T2, _T3, _T4, _T5]]: ...
 172     @overload
 173     def gather(
 174         coro_or_future1: _FutureT[Any],
 175         coro_or_future2: _FutureT[Any],
 176         coro_or_future3: _FutureT[Any],
 177         coro_or_future4: _FutureT[Any],
 178         coro_or_future5: _FutureT[Any],
 179         coro_or_future6: _FutureT[Any],
 180         *coros_or_futures: _FutureT[Any],
 181         loop: AbstractEventLoop | None = ...,
 182         return_exceptions: bool = ...,
 183     ) -> Future[list[Any]]: ...
 184     @overload
 185     def gather(
 186         coro_or_future1: _FutureT[_T1], *, loop: AbstractEventLoop | None = ..., return_exceptions: bool = ...
 187     ) -> Future[tuple[_T1 | BaseException]]: ...
 188     @overload
 189     def gather(
 190         coro_or_future1: _FutureT[_T1],
 191         coro_or_future2: _FutureT[_T2],
 192         *,
 193         loop: AbstractEventLoop | None = ...,
 194         return_exceptions: bool = ...,
 195     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException]]: ...
 196     @overload
 197     def gather(
 198         coro_or_future1: _FutureT[_T1],
 199         coro_or_future2: _FutureT[_T2],
 200         coro_or_future3: _FutureT[_T3],
 201         *,
 202         loop: AbstractEventLoop | None = ...,
 203         return_exceptions: bool = ...,
 204     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException]]: ...
 205     @overload
 206     def gather(
 207         coro_or_future1: _FutureT[_T1],
 208         coro_or_future2: _FutureT[_T2],
 209         coro_or_future3: _FutureT[_T3],
 210         coro_or_future4: _FutureT[_T4],
 211         *,
 212         loop: AbstractEventLoop | None = ...,
 213         return_exceptions: bool = ...,
 214     ) -> Future[tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException]]: ...
 215     @overload
 216     def gather(
 217         coro_or_future1: _FutureT[_T1],
 218         coro_or_future2: _FutureT[_T2],
 219         coro_or_future3: _FutureT[_T3],
 220         coro_or_future4: _FutureT[_T4],
 221         coro_or_future5: _FutureT[_T5],
 222         *,
 223         loop: AbstractEventLoop | None = ...,
 224         return_exceptions: bool = ...,
 225     ) -> Future[
 226         tuple[_T1 | BaseException, _T2 | BaseException, _T3 | BaseException, _T4 | BaseException, _T5 | BaseException]
 227     ]: ...
 228
 229 def run_coroutine_threadsafe(coro: _FutureT[_T], loop: AbstractEventLoop) -> concurrent.futures.Future[_T]: ...
 230
 231 if sys.version_info >= (3, 10):
 232     def shield(arg: _FutureT[_T]) -> Future[_T]: ...
 233     def sleep(delay: float, result: _T = ...) -> Future[_T]: ...
 234     @overload
 235     def wait(fs: Iterable[_FT], *, timeout: float | None = ..., return_when: str = ...) -> Future[tuple[set[_FT], set[_FT]]]: ...  # type: ignore
 236     @overload
 237     def wait(
 238         fs: Iterable[Awaitable[_T]], *, timeout: float | None = ..., return_when: str = ...
 239     ) -> Future[tuple[set[Task[_T]], set[Task[_T]]]]: ...
 240     def wait_for(fut: _FutureT[_T], timeout: float | None) -> Future[_T]: ...
 241
 242 else:
 243     def shield(arg: _FutureT[_T], *, loop: AbstractEventLoop | None = ...) -> Future[_T]: ...
 244     def sleep(delay: float, result: _T = ..., *, loop: AbstractEventLoop | None = ...) -> Future[_T]: ...
 245     @overload
 246     def wait(fs: Iterable[_FT], *, loop: AbstractEventLoop | None = ..., timeout: float | None = ..., return_when: str = ...) -> Future[tuple[set[_FT], set[_FT]]]: ...  # type: ignore
 247     @overload
 248     def wait(
 249         fs: Iterable[Awaitable[_T]], *, loop: AbstractEventLoop | None = ..., timeout: float | None = ..., return_when: str = ...
 250     ) -> Future[tuple[set[Task[_T]], set[Task[_T]]]]: ...
 251     def wait_for(fut: _FutureT[_T], timeout: float | None, *, loop: AbstractEventLoop | None = ...) -> Future[_T]: ...
 252
 253 class Task(Future[_T], Generic[_T]):
 254     if sys.version_info >= (3, 8):
 255         def __init__(
 256             self,
 257             coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T],
 258             *,
 259             loop: AbstractEventLoop = ...,
 260             name: str | None = ...,
 261         ) -> None: ...
 262     else:
 263         def __init__(
 264             self, coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T], *, loop: AbstractEventLoop = ...
 265         ) -> None: ...
 266     def __repr__(self) -> str: ...
 267     if sys.version_info >= (3, 8):
 268         def get_coro(self) -> Generator[_TaskYieldType, None, _T] | Awaitable[_T]: ...
 269         def get_name(self) -> str: ...
 270         def set_name(self, __value: object) -> None: ...
 271     def get_stack(self, *, limit: int | None = ...) -> list[FrameType]: ...
 272     def print_stack(self, *, limit: int | None = ..., file: TextIO | None = ...) -> None: ...
 273     if sys.version_info >= (3, 9):
 274         def cancel(self, msg: Any | None = ...) -> bool: ...
 275     else:
 276         def cancel(self) -> bool: ...
 277     if sys.version_info < (3, 9):
 278         @classmethod
 279         def current_task(cls, loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...
 280         @classmethod
 281         def all_tasks(cls, loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
 282     if sys.version_info < (3, 7):
 283         def _wakeup(self, fut: Future[Any]) -> None: ...
 284     if sys.version_info >= (3, 9):
 285         def __class_getitem__(cls, item: Any) -> GenericAlias: ...
 286
 287 if sys.version_info >= (3, 7):
 288     def all_tasks(loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
 289     if sys.version_info >= (3, 8):
 290         def create_task(coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T], *, name: str | None = ...) -> Task[_T]: ...
 291     else:
 292         def create_task(coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T]) -> Task[_T]: ...
 293     def current_task(loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...