VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-07 04:34:28 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fixes for asyncio (#654)

Browse Source 
* Fix signature of gather().  Also fold long lines and kill _GatheringFuture.

* Add several missing exports from tasks.
This commit is contained in:
Guido van Rossum
2016-11-04 11:19:22 -07:00
committed by GitHub
parent 7c20adb715
commit 7f7e2d4e75
2 changed files with 28 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
stdlib/3.4/asyncio/__init__.pyi
View File

@@ -39,13 +39,18 @@ from asyncio.futures import (
Future as Future,
)
from asyncio.tasks import (
sleep as sleep,
Task as Task,
FIRST_COMPLETED as FIRST_COMPLETED,
FIRST_EXCEPTION as FIRST_EXCEPTION,
ALL_COMPLETED as ALL_COMPLETED,
as_completed as as_completed,
ensure_future as ensure_future,
gather as gather,
run_coroutine_threadsafe as run_coroutine_threadsafe,
shield as shield,
sleep as sleep,
wait as wait,
wait_for as wait_for,
Task as Task,
)
from asyncio.events import (
AbstractEventLoopPolicy as AbstractEventLoopPolicy,

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

@@ -1,27 +1,33 @@
from typing import Any, TypeVar, Set, Dict, List, TextIO, Union, Tuple, Generic, Callable, Generator, Iterable, Awaitable, overload, Sequence, Iterator, Optional
from typing import (Any, TypeVar, Set, Dict, List, TextIO, Union, Tuple, Generic, Callable,
Generator, Iterable, Awaitable, overload, Sequence, Iterator, Optional)
__all__ = ... # type: str
from .events import AbstractEventLoop
from .futures import Future
_T = TypeVar('_T')
FIRST_EXCEPTION = 'FIRST_EXCEPTION'
FIRST_COMPLETED = 'FIRST_COMPLETED'
ALL_COMPLETED = 'ALL_COMPLETED'
_T = TypeVar('_T')
def as_completed(fs: Sequence[Future[_T]], *, loop: AbstractEventLoop = ..., timeout=None) -> Iterator[Generator[Any, None, _T]]: ...
def ensure_future(coro_or_future: Union[Future[_T], Generator[Any, None, _T]], *, loop: AbstractEventLoop = ...) -> Future[_T]: ...
def gather(*coros_or_futures: Sequence[Union[Future[_T], Generator[Any, None, _T]]], loop: AbstractEventLoop = ..., return_exceptions: bool = False) -> Future[_T]: ...
def run_coroutine_threadsafe(coro: Generator[Any, None, _T], loop: AbstractEventLoop) -> Future[_T]: ...
def shield(arg: Union[Future[_T], Generator[Any, None, _T]], *, loop: AbstractEventLoop = ...) -> Future[_T]: ...
def as_completed(fs: Sequence[Future[_T]], *, loop: AbstractEventLoop = ...,
timeout=None) -> Iterator[Generator[Any, None, _T]]: ...
def ensure_future(coro_or_future: Union[Future[_T], Generator[Any, None, _T]],
*, loop: AbstractEventLoop = ...) -> Future[_T]: ...
def gather(*coros_or_futures: Union[Future[_T], Generator[Any, None, _T], Awaitable[_T]],
loop: AbstractEventLoop = ..., return_exceptions: bool = False) -> Future[List[_T]]: ...
def run_coroutine_threadsafe(coro: Generator[Any, None, _T],
loop: AbstractEventLoop) -> Future[_T]: ...
def shield(arg: Union[Future[_T], Generator[Any, None, _T]],
*, loop: AbstractEventLoop = ...) -> Future[_T]: ...
def sleep(delay: float, result: _T = ..., loop: AbstractEventLoop = ...) -> Future[_T]: ...
def wait(fs: List[Task[_T]], *, loop: AbstractEventLoop = ...,
timeout: float = ..., return_when: str = ...) -> Future[Tuple[Set[Future[_T]], Set[Future[_T]]]]: ...
def wait_for(fut: Union[Future[_T], Generator[Any, None, _T]], timeout: Optional[float], *, loop: AbstractEventLoop = ...) -> Future[_T]: ...
class _GatheringFuture(Future[_T], Generic[_T]):
def __init__(self, children: Sequence[Union[Future[_T], Generator[Any, None, _T]]], *, loop: AbstractEventLoop = ...) -> None: ...
def cancel(self) -> bool: ...
timeout: float = ...,
return_when: str = ...) -> Future[Tuple[Set[Future[_T]], Set[Future[_T]]]]: ...
def wait_for(fut: Union[Future[_T], Generator[Any, None, _T]], timeout: Optional[float],
*, loop: AbstractEventLoop = ...) -> Future[_T]: ...
class Task(Future[_T], Generic[_T]):
_all_tasks = ... # type: Set[Task]
@@ -33,7 +39,8 @@ class Task(Future[_T], Generic[_T]):
# Can't use a union, see mypy issue #1873.
@overload
def __init__(self, coro: Generator[Any, None, _T], *, loop: AbstractEventLoop = ...) -> None: ...
def __init__(self, coro: Generator[Any, None, _T],
*, loop: AbstractEventLoop = ...) -> None: ...
@overload
def __init__(self, coro: Awaitable[_T], *, loop: AbstractEventLoop = ...) -> None: ...