Format: application/octet-stream
⬇ Download File (895 B)
   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