-
-
Notifications
You must be signed in to change notification settings - Fork 2k
Expand file tree
/
Copy path_base.pyi
More file actions
119 lines (102 loc) · 4.24 KB
/
_base.pyi
File metadata and controls
119 lines (102 loc) · 4.24 KB
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import sys
import threading
from _typeshed import Unused
from collections.abc import Callable, Iterable, Iterator
from logging import Logger
from types import GenericAlias, TracebackType
from typing import Any, Final, Generic, NamedTuple, Protocol, TypeVar, type_check_only
from typing_extensions import ParamSpec, Self
FIRST_COMPLETED: Final = "FIRST_COMPLETED"
FIRST_EXCEPTION: Final = "FIRST_EXCEPTION"
ALL_COMPLETED: Final = "ALL_COMPLETED"
PENDING: Final = "PENDING"
RUNNING: Final = "RUNNING"
CANCELLED: Final = "CANCELLED"
CANCELLED_AND_NOTIFIED: Final = "CANCELLED_AND_NOTIFIED"
FINISHED: Final = "FINISHED"
_STATE_TO_DESCRIPTION_MAP: Final[dict[str, str]]
LOGGER: Logger
class Error(Exception): ...
class CancelledError(Error): ...
if sys.version_info >= (3, 11):
from builtins import TimeoutError as TimeoutError
else:
class TimeoutError(Error): ...
class InvalidStateError(Error): ...
class BrokenExecutor(RuntimeError): ...
_T = TypeVar("_T")
_T_co = TypeVar("_T_co", covariant=True)
_P = ParamSpec("_P")
class Future(Generic[_T]):
_condition: threading.Condition
_state: str
_result: _T | None
_exception: BaseException | None
_waiters: list[_Waiter]
def cancel(self) -> bool: ...
def cancelled(self) -> bool: ...
def running(self) -> bool: ...
def done(self) -> bool: ...
def add_done_callback(self, fn: Callable[[Future[_T]], object]) -> None: ...
def result(self, timeout: float | None = None) -> _T: ...
def set_running_or_notify_cancel(self) -> bool: ...
def set_result(self, result: _T) -> None: ...
def exception(self, timeout: float | None = None) -> BaseException | None: ...
def set_exception(self, exception: BaseException | None) -> None: ...
def __class_getitem__(cls, item: Any, /) -> GenericAlias: ...
class Executor:
def submit(self, fn: Callable[_P, _T], /, *args: _P.args, **kwargs: _P.kwargs) -> Future[_T]: ...
if sys.version_info >= (3, 14):
def map(
self,
fn: Callable[..., _T],
*iterables: Iterable[Any],
timeout: float | None = None,
chunksize: int = 1,
buffersize: int | None = None,
) -> Iterator[_T]: ...
else:
def map(
self, fn: Callable[..., _T], *iterables: Iterable[Any], timeout: float | None = None, chunksize: int = 1
) -> Iterator[_T]: ...
def shutdown(self, wait: bool = True, *, cancel_futures: bool = False) -> None: ...
def __enter__(self) -> Self: ...
def __exit__(
self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
) -> bool | None: ...
@type_check_only
class _AsCompletedFuture(Protocol[_T_co]):
# as_completed only mutates non-generic aspects of passed Futures and does not do any nominal
# checks. Therefore, we can use a Protocol here to allow as_completed to act covariantly.
# See the tests for concurrent.futures
_condition: threading.Condition
_state: str
_waiters: list[_Waiter]
# Not used by as_completed, but needed to propagate the generic type
def result(self, timeout: float | None = None) -> _T_co: ...
def as_completed(fs: Iterable[_AsCompletedFuture[_T]], timeout: float | None = None) -> Iterator[Future[_T]]: ...
class DoneAndNotDoneFutures(NamedTuple, Generic[_T]):
done: set[Future[_T]]
not_done: set[Future[_T]]
def wait(
fs: Iterable[Future[_T]], timeout: float | None = None, return_when: str = "ALL_COMPLETED"
) -> DoneAndNotDoneFutures[_T]: ...
class _Waiter:
event: threading.Event
finished_futures: list[Future[Any]]
def add_result(self, future: Future[Any]) -> None: ...
def add_exception(self, future: Future[Any]) -> None: ...
def add_cancelled(self, future: Future[Any]) -> None: ...
class _AsCompletedWaiter(_Waiter):
lock: threading.Lock
class _FirstCompletedWaiter(_Waiter): ...
class _AllCompletedWaiter(_Waiter):
num_pending_calls: int
stop_on_exception: bool
lock: threading.Lock
def __init__(self, num_pending_calls: int, stop_on_exception: bool) -> None: ...
class _AcquireFutures:
futures: Iterable[Future[Any]]
def __init__(self, futures: Iterable[Future[Any]]) -> None: ...
def __enter__(self) -> None: ...
def __exit__(self, *args: Unused) -> None: ...