Format: application/octet-stream
1 from collections import deque
2 from typing import Any, Deque, Generic, Optional, TypeVar
3
4 _T = TypeVar("_T")
5
6 class Empty(Exception): ...
7 class Full(Exception): ...
8
9 class Queue(Generic[_T]):
10 maxsize: Any
11 mutex: Any
12 not_empty: Any
13 not_full: Any
14 all_tasks_done: Any
15 unfinished_tasks: Any
16 queue: Deque[Any] # undocumented
17 def __init__(self, maxsize: int = ...) -> None: ...
18 def task_done(self) -> None: ...
19 def join(self) -> None: ...
20 def qsize(self) -> int: ...
21 def empty(self) -> bool: ...
22 def full(self) -> bool: ...
23 def put(self, item: _T, block: bool = ..., timeout: Optional[float] = ...) -> None: ...
24 def put_nowait(self, item: _T) -> None: ...
25 def get(self, block: bool = ..., timeout: Optional[float] = ...) -> _T: ...
26 def get_nowait(self) -> _T: ...
27
28 class PriorityQueue(Queue[_T]): ...
29 class LifoQueue(Queue[_T]): ...
30