VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-10 05:51:52 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Improve multiprocessing.context module (#8466)

Browse Source 
* Use `ClassVar` where applicable
* Fix `_Popen()` return types
* Remove non-existing field `BaseContext.Process`

Co-authored-by: Alex Waygood <Alex.Waygood@Gmail.com>
This commit is contained in:
Nikita Sobolev
2022-08-04 13:30:23 +03:00
committed by GitHub
parent bcd770bad3
commit 799fd2c8d8
1 changed files with 14 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
stdlib/multiprocessing/context.pyi
View File

@@ -1,10 +1,9 @@
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, synchronize
from multiprocessing import popen_fork, popen_forkserver, popen_spawn_posix, popen_spawn_win32, queues, synchronize
from multiprocessing.connection import _ConnectionBase
from multiprocessing.managers import SyncManager
from multiprocessing.pool import Pool as _Pool
@@ -27,11 +26,10 @@ class TimeoutError(ProcessError): ...
class AuthenticationError(ProcessError): ...
class BaseContext:
Process: type[BaseProcess]
ProcessError: type[Exception]
BufferTooShort: type[Exception]
TimeoutError: type[Exception]
AuthenticationError: type[Exception]
ProcessError: ClassVar[type[ProcessError]]
BufferTooShort: ClassVar[type[BufferTooShort]]
TimeoutError: ClassVar[type[TimeoutError]]
AuthenticationError: ClassVar[type[AuthenticationError]]
# N.B. The methods below are applied at runtime to generate
# multiprocessing.*, so the signatures should be identical (modulo self).
@@ -137,7 +135,7 @@ class Process(BaseProcess):
def _Popen(process_obj: BaseProcess) -> DefaultContext: ...
class DefaultContext(BaseContext):
Process: type[multiprocessing.Process]
Process: ClassVar[type[Process]]
def __init__(self, context: BaseContext) -> None: ...
def set_start_method(self, method: str | None, force: bool = ...) -> None: ...
def get_start_method(self, allow_none: bool = ...) -> str: ...
@@ -151,39 +149,39 @@ if sys.platform != "win32":
class ForkProcess(BaseProcess):
_start_method: str
@staticmethod
def _Popen(process_obj: BaseProcess) -> Any: ...
def _Popen(process_obj: BaseProcess) -> popen_fork.Popen: ...
class SpawnProcess(BaseProcess):
_start_method: str
@staticmethod
def _Popen(process_obj: BaseProcess) -> SpawnProcess: ...
def _Popen(process_obj: BaseProcess) -> popen_spawn_posix.Popen: ...
class ForkServerProcess(BaseProcess):
_start_method: str
@staticmethod
def _Popen(process_obj: BaseProcess) -> Any: ...
def _Popen(process_obj: BaseProcess) -> popen_forkserver.Popen: ...
class ForkContext(BaseContext):
_name: str
Process: type[ForkProcess]
Process: ClassVar[type[ForkProcess]]
class SpawnContext(BaseContext):
_name: str
Process: type[SpawnProcess]
Process: ClassVar[type[SpawnProcess]]
class ForkServerContext(BaseContext):
_name: str
Process: type[ForkServerProcess]
Process: ClassVar[type[ForkServerProcess]]
else:
class SpawnProcess(BaseProcess):
_start_method: str
@staticmethod
def _Popen(process_obj: BaseProcess) -> Any: ...
def _Popen(process_obj: BaseProcess) -> popen_spawn_win32.Popen: ...
class SpawnContext(BaseContext):
_name: str
Process: type[SpawnProcess]
Process: ClassVar[type[SpawnProcess]]
def _force_start_method(method: str) -> None: ...
def get_spawning_popen() -> Any | None: ...