Use TypeAlias where possible for type aliases (#7630)

This commit is contained in:
Alex Waygood
2022-04-16 02:01:00 +01:00
committed by GitHub
parent c0e6dd3f3f
commit 740193a8fc
218 changed files with 760 additions and 625 deletions

View File

@@ -20,8 +20,8 @@ from multiprocessing.process import active_children as active_children, current_
# multiprocessing.queues or the aliases defined below. See #4266 for discussion.
from multiprocessing.queues import JoinableQueue as JoinableQueue, Queue as Queue, SimpleQueue as SimpleQueue
from multiprocessing.spawn import freeze_support as freeze_support
from typing import Any, overload
from typing_extensions import Literal
from typing import Any, TypeVar, overload
from typing_extensions import Literal, TypeAlias
if sys.version_info >= (3, 8):
from multiprocessing.process import parent_process as parent_process
@@ -118,23 +118,24 @@ else:
# from multiprocessing import _LockType
# lock: _LockType = Lock()
_QueueType = Queue
_SimpleQueueType = SimpleQueue
_JoinableQueueType = JoinableQueue
_BarrierType = synchronize.Barrier
_BoundedSemaphoreType = synchronize.BoundedSemaphore
_ConditionType = synchronize.Condition
_EventType = synchronize.Event
_LockType = synchronize.Lock
_RLockType = synchronize.RLock
_SemaphoreType = synchronize.Semaphore
_T = TypeVar("_T")
_QueueType: TypeAlias = Queue[_T]
_SimpleQueueType: TypeAlias = SimpleQueue[_T]
_JoinableQueueType: TypeAlias = JoinableQueue[_T]
_BarrierType: TypeAlias = synchronize.Barrier
_BoundedSemaphoreType: TypeAlias = synchronize.BoundedSemaphore
_ConditionType: TypeAlias = synchronize.Condition
_EventType: TypeAlias = synchronize.Event
_LockType: TypeAlias = synchronize.Lock
_RLockType: TypeAlias = synchronize.RLock
_SemaphoreType: TypeAlias = synchronize.Semaphore
# N.B. The functions below are generated at runtime by partially applying
# multiprocessing.context.BaseContext's methods, so the two signatures should
# be identical (modulo self).
# Synchronization primitives
_LockLike = synchronize.Lock | synchronize.RLock
_LockLike: TypeAlias = synchronize.Lock | synchronize.RLock
RawValue = context._default_context.RawValue
RawArray = context._default_context.RawArray
Value = context._default_context.Value

View File

@@ -3,12 +3,12 @@ import sys
import types
from _typeshed import Self
from typing import Any, Iterable, Union
from typing_extensions import SupportsIndex
from typing_extensions import SupportsIndex, TypeAlias
__all__ = ["Client", "Listener", "Pipe", "wait"]
# https://docs.python.org/3/library/multiprocessing.html#address-formats
_Address = Union[str, tuple[str, int]]
_Address: TypeAlias = Union[str, tuple[str, int]]
class _ConnectionBase:
def __init__(self, handle: SupportsIndex, readable: bool = ..., writable: bool = ...) -> None: ...

View File

@@ -9,14 +9,14 @@ from multiprocessing.pool import Pool as _Pool
from multiprocessing.process import BaseProcess
from multiprocessing.sharedctypes import SynchronizedArray, SynchronizedBase
from typing import Any, ClassVar, TypeVar, overload
from typing_extensions import Literal
from typing_extensions import Literal, TypeAlias
if sys.version_info >= (3, 8):
__all__ = ()
else:
__all__: list[str] = []
_LockLike = synchronize.Lock | synchronize.RLock
_LockLike: TypeAlias = synchronize.Lock | synchronize.RLock
_CT = TypeVar("_CT", bound=_CData)
class ProcessError(Exception): ...

View File

@@ -2,12 +2,13 @@ from _typeshed import Self
from queue import Queue
from types import TracebackType
from typing import Any, Union
from typing_extensions import TypeAlias
__all__ = ["Client", "Listener", "Pipe"]
families: list[None]
_Address = Union[str, tuple[str, int]]
_Address: TypeAlias = Union[str, tuple[str, int]]
class Connection:
_in: Any

View File

@@ -4,10 +4,11 @@ from contextlib import AbstractContextManager
from multiprocessing.context import BaseContext
from types import TracebackType
from typing import Any, Callable
from typing_extensions import TypeAlias
__all__ = ["Lock", "RLock", "Semaphore", "BoundedSemaphore", "Condition", "Event"]
_LockLike = Lock | RLock
_LockLike: TypeAlias = Lock | RLock
class Barrier(threading.Barrier):
def __init__(