Add type stubs for multiprocessing.Manager() (#1270)

* Use Dict/List types and _Namespace alias

* Use TypeVar definitions for mapping/sequence stubs
This commit is contained in:
John Reese
2017-05-22 15:31:23 -07:00
committed by Jelle Zijlstra
parent b6d08b81a3
commit 54408054cc
2 changed files with 27 additions and 1 deletions

View File

@@ -3,6 +3,7 @@
from typing import Any, Callable, Iterable, Mapping, Optional, Dict, List
from multiprocessing.context import BaseContext
from multiprocessing.managers import SyncManager
from multiprocessing.process import current_process as current_process
class Lock():
@@ -105,3 +106,4 @@ class Value():
# ----- multiprocessing function stubs -----
def cpu_count() -> int: ...
def freeze_support() -> None: ...
def Manager() -> SyncManager: ...

View File

@@ -2,9 +2,33 @@
# NOTE: These are incomplete!
from typing import Any
import queue
import threading
from typing import Any, Dict, List, Mapping, Sequence, TypeVar
_T = TypeVar('_T')
_KT = TypeVar('_KT')
_VT = TypeVar('_VT')
class Namespace: ...
_Namespace = Namespace
class BaseManager:
def register(self, typeid: str, callable: Any = ...) -> None: ...
class SyncManager(BaseManager):
def BoundedSemaphore(self, value: Any = ...) -> threading.BoundedSemaphore: ...
def Condition(self, lock: Any = ...) -> threading.Condition: ...
def Event(self) -> threading.Event: ...
def Lock(self) -> threading.Lock: ...
def Namespace(self) -> _Namespace: ...
def Queue(self, maxsize: int = ...) -> queue.Queue: ...
def RLock(self) -> threading.RLock: ...
def Semaphore(self, value: Any = ...) -> threading.Semaphore: ...
def Array(self, typecode: Any, sequence: Sequence[_T]) -> Sequence[_T]: ...
def Value(self, typecode: Any, value: _T) -> _T: ...
def dict(self, sequence: Mapping[_KT, _VT] = ...) -> Dict[_KT, _VT]: ...
def list(self, sequence: Sequence[_T] = ...) -> List[_T]: ...
class RemoteError(Exception): ...