VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-09 21:46:42 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add multiprocessing.sharedctypes stubs (#5231)

Browse Source
This commit is contained in:
hatal175
2021-05-02 23:20:33 +03:00
committed by GitHub
parent 06709912ad
commit b15c025059
3 changed files with 152 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
stdlib/multiprocessing/context.pyi
View File

@@ -1,12 +1,17 @@
import ctypes
import multiprocessing
import sys
from collections.abc import Callable, Iterable, Sequence
from ctypes import _CData
from logging import Logger
from multiprocessing import queues, sharedctypes, synchronize
from multiprocessing import queues, synchronize
from multiprocessing.process import BaseProcess
from typing import Any, Callable, Iterable, List, Optional, Sequence, Type, Union, overload
from multiprocessing.sharedctypes import SynchronizedArray, SynchronizedBase
from typing import Any, Optional, Type, TypeVar, Union, overload
from typing_extensions import Literal
_LockLike = Union[synchronize.Lock, synchronize.RLock]
_CT = TypeVar("_CT", bound=_CData)
class ProcessError(Exception): ...
class BufferTooShort(ProcessError): ...
@@ -28,7 +33,7 @@ class BaseContext(object):
@staticmethod
def parent_process() -> Optional[BaseProcess]: ...
@staticmethod
def active_children() -> List[BaseProcess]: ...
def active_children() -> list[BaseProcess]: ...
def cpu_count(self) -> int: ...
# TODO: change return to SyncManager once a stub exists in multiprocessing.managers
def Manager(self) -> Any: ...
@@ -53,28 +58,52 @@ class BaseContext(object):
initargs: Iterable[Any] = ...,
maxtasksperchild: Optional[int] = ...,
) -> multiprocessing.pool.Pool: ...
# TODO: typecode_or_type param is a ctype with a base class of _SimpleCData or array.typecode Need to figure out
# how to handle the ctype
# TODO: change return to RawValue once a stub exists in multiprocessing.sharedctypes
def RawValue(self, typecode_or_type: Any, *args: Any) -> Any: ...
# TODO: typecode_or_type param is a ctype with a base class of _SimpleCData or array.typecode Need to figure out
# how to handle the ctype
# TODO: change return to RawArray once a stub exists in multiprocessing.sharedctypes
def RawArray(self, typecode_or_type: Any, size_or_initializer: Union[int, Sequence[Any]]) -> Any: ...
# TODO: typecode_or_type param is a ctype with a base class of _SimpleCData or array.typecode Need to figure out
# how to handle the ctype
def Value(self, typecode_or_type: Any, *args: Any, lock: bool = ...) -> sharedctypes._Value: ...
# TODO: typecode_or_type param is a ctype with a base class of _SimpleCData or array.typecode Need to figure out
# how to handle the ctype
@overload
def RawValue(self, typecode_or_type: Type[_CT], *args: Any) -> _CT: ...
@overload
def RawValue(self, typecode_or_type: str, *args: Any) -> Any: ...
@overload
def RawArray(self, typecode_or_type: Type[_CT], size_or_initializer: Union[int, Sequence[Any]]) -> ctypes.Array[_CT]: ...
@overload
def RawArray(self, typecode_or_type: str, size_or_initializer: Union[int, Sequence[Any]]) -> Any: ...
@overload
def Value(self, typecode_or_type: Type[_CT], *args: Any, lock: Literal[False]) -> _CT: ...
@overload
def Value(self, typecode_or_type: Type[_CT], *args: Any, lock: Union[Literal[True], _LockLike]) -> SynchronizedBase[_CT]: ...
@overload
def Value(self, typecode_or_type: str, *args: Any, lock: Union[Literal[True], _LockLike]) -> SynchronizedBase[Any]: ...
@overload
def Value(self, typecode_or_type: Union[str, Type[_CData]], *args: Any, lock: Union[bool, _LockLike] = ...) -> Any: ...
@overload
def Array(
self, typecode_or_type: Any, size_or_initializer: Union[int, Sequence[Any]], *, lock: bool = ...
) -> sharedctypes._Array: ...
self, typecode_or_type: Type[_CT], size_or_initializer: Union[int, Sequence[Any]], *, lock: Literal[False]
) -> _CT: ...
@overload
def Array(
self,
typecode_or_type: Type[_CT],
size_or_initializer: Union[int, Sequence[Any]],
*,
lock: Union[Literal[True], _LockLike],
) -> SynchronizedArray[_CT]: ...
@overload
def Array(
self, typecode_or_type: str, size_or_initializer: Union[int, Sequence[Any]], *, lock: Union[Literal[True], _LockLike]
) -> SynchronizedArray[Any]: ...
@overload
def Array(
self,
typecode_or_type: Union[str, Type[_CData]],
size_or_initializer: Union[int, Sequence[Any]],
*,
lock: Union[bool, _LockLike] = ...,
) -> Any: ...
def freeze_support(self) -> None: ...
def get_logger(self) -> Logger: ...
def log_to_stderr(self, level: Optional[str] = ...) -> Logger: ...
def allow_connection_pickling(self) -> None: ...
def set_executable(self, executable: str) -> None: ...
def set_forkserver_preload(self, module_names: List[str]) -> None: ...
def set_forkserver_preload(self, module_names: list[str]) -> None: ...
if sys.platform != "win32":
@overload
def get_context(self, method: None = ...) -> DefaultContext: ...
@@ -111,7 +140,9 @@ class DefaultContext(BaseContext):
def __init__(self, context: BaseContext) -> None: ...
def set_start_method(self, method: Optional[str], force: bool = ...) -> None: ...
def get_start_method(self, allow_none: bool = ...) -> str: ...
def get_all_start_methods(self) -> List[str]: ...
def get_all_start_methods(self) -> list[str]: ...
_default_context: DefaultContext
if sys.platform != "win32":
class ForkProcess(BaseProcess):