VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-09 05:24:52 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

More precisely typehint asyncio.gather up to 5 params (#1550)

Browse Source
This commit is contained in:
Liran Nuna
2017-08-28 21:37:51 -07:00
committed by Jelle Zijlstra
parent 258b9fb821
commit c5f1e907fa
1 changed files with 24 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
stdlib/3.4/asyncio/tasks.pyi
View File

@@ -8,6 +8,11 @@ from .futures import Future
__all__: List[str]
_T = TypeVar('_T')
_T1 = TypeVar('_T1')
_T2 = TypeVar('_T2')
_T3 = TypeVar('_T3')
_T4 = TypeVar('_T4')
_T5 = TypeVar('_T5')
_FutureT = Union[Future[_T], Generator[Any, None, _T], Awaitable[_T]]
FIRST_EXCEPTION = 'FIRST_EXCEPTION'
@@ -19,10 +24,25 @@ def as_completed(fs: Sequence[_FutureT[_T]], *, loop: AbstractEventLoop = ...,
def ensure_future(coro_or_future: _FutureT[_T],
*, loop: AbstractEventLoop = ...) -> Future[_T]: ...
async = ensure_future
# TODO: gather() should use variadic type vars instead of _TAny.
_TAny = Any
def gather(*coros_or_futures: _FutureT[_TAny],
loop: AbstractEventLoop = ..., return_exceptions: bool = False) -> Future[List[_TAny]]: ...
@overload
def gather(coro_or_future1: _FutureT[_T1],
*, loop: AbstractEventLoop = ..., return_exceptions: bool = False) -> Future[Tuple[_T1]]: ...
@overload
def gather(coro_or_future1: _FutureT[_T1], coro_or_future2: _FutureT[_T2],
*, loop: AbstractEventLoop = ..., return_exceptions: bool = False) -> Future[Tuple[_T1, _T2]]: ...
@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 = ..., return_exceptions: bool = 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 = ..., return_exceptions: bool = 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 = ..., return_exceptions: bool = False) -> Future[Tuple[Any, ...]]: ...
def run_coroutine_threadsafe(coro: _FutureT[_T],
loop: AbstractEventLoop) -> concurrent.futures.Future[_T]: ...
def shield(arg: _FutureT[_T], *, loop: AbstractEventLoop = ...) -> Future[_T]: ...