.config/coc/extensions/coc-python-data/languageServer.0.5.59/Typeshed/stdlib/3/multiprocessing/managers.pyi

   1 # Stubs for multiprocessing.managers\r
   2 \r
   3 # NOTE: These are incomplete!\r
   4 \r
   5 import queue\r
   6 import threading\r
   7 from typing import (\r
   8     Any, Callable, ContextManager, Dict, Iterable, List, Mapping, Optional,\r
   9     Sequence, Tuple, TypeVar, Union,\r
  10 )\r
  11 \r
  12 _T = TypeVar('_T')\r
  13 _KT = TypeVar('_KT')\r
  14 _VT = TypeVar('_VT')\r
  15 \r
  16 class Namespace: ...\r
  17 \r
  18 _Namespace = Namespace\r
  19 \r
  20 class BaseManager(ContextManager[BaseManager]):\r
  21     address: Union[str, Tuple[str, int]]\r
  22     def connect(self) -> None: ...\r
  23     def register(self, typeid: str, callable: Any = ...) -> None: ...\r
  24     def shutdown(self) -> None: ...\r
  25     def start(self, initializer: Optional[Callable[..., Any]] = ...,\r
  26               initargs: Iterable[Any] = ...) -> None: ...\r
  27 \r
  28 class SyncManager(BaseManager):\r
  29     def BoundedSemaphore(self, value: Any = ...) -> threading.BoundedSemaphore: ...\r
  30     def Condition(self, lock: Any = ...) -> threading.Condition: ...\r
  31     def Event(self) -> threading.Event: ...\r
  32     def Lock(self) -> threading.Lock: ...\r
  33     def Namespace(self) -> _Namespace: ...\r
  34     def Queue(self, maxsize: int = ...) -> queue.Queue: ...\r
  35     def RLock(self) -> threading.RLock: ...\r
  36     def Semaphore(self, value: Any = ...) -> threading.Semaphore: ...\r
  37     def Array(self, typecode: Any, sequence: Sequence[_T]) -> Sequence[_T]: ...\r
  38     def Value(self, typecode: Any, value: _T) -> _T: ...\r
  39     def dict(self, sequence: Mapping[_KT, _VT] = ...) -> Dict[_KT, _VT]: ...\r
  40     def list(self, sequence: Sequence[_T] = ...) -> List[_T]: ...\r
  41 \r
  42 class RemoteError(Exception): ...\r