-
-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
futures.pyi
62 lines (55 loc) · 2.3 KB
/
futures.pyi
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import sys
from typing import Any, Union, Callable, TypeVar, Type, List, Iterable, Generator, Awaitable, Optional, Tuple
from .events import AbstractEventLoop
from concurrent.futures import (
Future as _ConcurrentFuture,
Error,
)
if sys.version_info < (3, 8):
from concurrent.futures import CancelledError as CancelledError
from concurrent.futures import TimeoutError as TimeoutError
class InvalidStateError(Error): ...
if sys.version_info >= (3, 7):
from contextvars import Context
_T = TypeVar('_T')
_S = TypeVar('_S')
class _TracebackLogger:
exc: BaseException
tb: List[str]
def __init__(self, exc: Any, loop: AbstractEventLoop) -> None: ...
def activate(self) -> None: ...
def clear(self) -> None: ...
def __del__(self) -> None: ...
def isfuture(obj: object) -> bool: ...
class Future(Awaitable[_T], Iterable[_T]):
_state: str
_exception: BaseException
_blocking = False
_log_traceback = False
_tb_logger: Type[_TracebackLogger]
def __init__(self, *, loop: Optional[AbstractEventLoop] = ...) -> None: ...
def __repr__(self) -> str: ...
def __del__(self) -> None: ...
if sys.version_info >= (3, 7):
def get_loop(self) -> AbstractEventLoop: ...
def _callbacks(self: _S) -> List[Tuple[Callable[[_S], Any], Context]]: ...
def add_done_callback(self: _S, __fn: Callable[[_S], Any], *, context: Optional[Context] = ...) -> None: ...
else:
@property
def _callbacks(self: _S) -> List[Callable[[_S], Any]]: ...
def add_done_callback(self: _S, __fn: Callable[[_S], Any]) -> None: ...
def cancel(self) -> bool: ...
def _schedule_callbacks(self) -> None: ...
def cancelled(self) -> bool: ...
def done(self) -> bool: ...
def result(self) -> _T: ...
def exception(self) -> BaseException: ...
def remove_done_callback(self: _S, fn: Callable[[_S], Any]) -> int: ...
def set_result(self, result: _T) -> None: ...
def set_exception(self, exception: Union[type, BaseException]) -> None: ...
def _copy_state(self, other: Any) -> None: ...
def __iter__(self) -> Generator[Any, None, _T]: ...
def __await__(self) -> Generator[Any, None, _T]: ...
@property
def _loop(self) -> AbstractEventLoop: ...
def wrap_future(f: Union[_ConcurrentFuture[_T], Future[_T]]) -> Future[_T]: ...