mirror of
https://github.com/davidhalter/typeshed.git
synced 2025-12-09 05:24:52 +08:00
Use PEP 604 syntax wherever possible (#7493)
This commit is contained in:
@@ -8,7 +8,7 @@ from asyncio.tasks import Task
|
||||
from asyncio.transports import BaseTransport, ReadTransport, SubprocessTransport, WriteTransport
|
||||
from collections.abc import Iterable
|
||||
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
|
||||
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, Union, overload
|
||||
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, overload
|
||||
from typing_extensions import Literal
|
||||
|
||||
if sys.version_info >= (3, 7):
|
||||
@@ -23,7 +23,7 @@ _ProtocolT = TypeVar("_ProtocolT", bound=BaseProtocol)
|
||||
_Context = dict[str, Any]
|
||||
_ExceptionHandler = Callable[[AbstractEventLoop, _Context], Any]
|
||||
_ProtocolFactory = Callable[[], BaseProtocol]
|
||||
_SSLContext = Union[bool, None, ssl.SSLContext]
|
||||
_SSLContext = bool | None | ssl.SSLContext
|
||||
|
||||
class Server(AbstractServer):
|
||||
if sys.version_info >= (3, 7):
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
import subprocess
|
||||
from collections import deque
|
||||
from typing import IO, Any, Callable, Optional, Sequence, Union
|
||||
from typing import IO, Any, Callable, Sequence
|
||||
|
||||
from . import events, futures, protocols, transports
|
||||
|
||||
_File = Optional[Union[int, IO[Any]]]
|
||||
_File = int | IO[Any] | None
|
||||
|
||||
class BaseSubprocessTransport(transports.SubprocessTransport):
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ import sys
|
||||
from _typeshed import FileDescriptorLike, Self
|
||||
from abc import ABCMeta, abstractmethod
|
||||
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
|
||||
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, Union, overload
|
||||
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, overload
|
||||
from typing_extensions import Literal
|
||||
|
||||
from .base_events import Server
|
||||
@@ -78,7 +78,7 @@ _ProtocolT = TypeVar("_ProtocolT", bound=BaseProtocol)
|
||||
_Context = dict[str, Any]
|
||||
_ExceptionHandler = Callable[[AbstractEventLoop, _Context], Any]
|
||||
_ProtocolFactory = Callable[[], BaseProtocol]
|
||||
_SSLContext = Union[bool, None, ssl.SSLContext]
|
||||
_SSLContext = bool | None | ssl.SSLContext
|
||||
|
||||
class Handle:
|
||||
_cancelled: bool
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
import functools
|
||||
import traceback
|
||||
from types import FrameType, FunctionType
|
||||
from typing import Any, Iterable, Union, overload
|
||||
from typing import Any, Iterable, overload
|
||||
|
||||
class _HasWrapper:
|
||||
__wrapper__: _HasWrapper | FunctionType
|
||||
|
||||
_FuncType = Union[FunctionType, _HasWrapper, functools.partial[Any], functools.partialmethod[Any]]
|
||||
_FuncType = FunctionType | _HasWrapper | functools.partial[Any] | functools.partialmethod[Any]
|
||||
|
||||
@overload
|
||||
def _get_function_source(func: _FuncType) -> tuple[str, int]: ...
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import sys
|
||||
from _typeshed import Self, StrPath
|
||||
from typing import Any, AsyncIterator, Awaitable, Callable, Iterable, Optional
|
||||
from typing import Any, AsyncIterator, Awaitable, Callable, Iterable
|
||||
|
||||
from . import events, protocols, transports
|
||||
from .base_events import Server
|
||||
@@ -64,7 +64,7 @@ else:
|
||||
"start_unix_server",
|
||||
]
|
||||
|
||||
_ClientConnectedCallback = Callable[[StreamReader, StreamWriter], Optional[Awaitable[None]]]
|
||||
_ClientConnectedCallback = Callable[[StreamReader, StreamWriter], Awaitable[None] | None]
|
||||
|
||||
if sys.version_info < (3, 8):
|
||||
class IncompleteReadError(EOFError):
|
||||
|
||||
@@ -2,7 +2,7 @@ import subprocess
|
||||
import sys
|
||||
from _typeshed import StrOrBytesPath
|
||||
from asyncio import events, protocols, streams, transports
|
||||
from typing import IO, Any, Callable, Union
|
||||
from typing import IO, Any, Callable
|
||||
from typing_extensions import Literal
|
||||
|
||||
if sys.version_info >= (3, 7):
|
||||
@@ -13,7 +13,7 @@ else:
|
||||
if sys.version_info >= (3, 8):
|
||||
_ExecArg = StrOrBytesPath
|
||||
else:
|
||||
_ExecArg = Union[str, bytes]
|
||||
_ExecArg = str | bytes
|
||||
|
||||
PIPE: int
|
||||
STDOUT: int
|
||||
|
||||
@@ -2,7 +2,7 @@ import concurrent.futures
|
||||
import sys
|
||||
from collections.abc import Awaitable, Generator, Iterable, Iterator
|
||||
from types import FrameType
|
||||
from typing import Any, Coroutine, Generic, Optional, TextIO, TypeVar, Union, overload
|
||||
from typing import Any, Coroutine, Generic, TextIO, TypeVar, overload
|
||||
from typing_extensions import Literal
|
||||
|
||||
from .events import AbstractEventLoop
|
||||
@@ -56,8 +56,8 @@ _T3 = TypeVar("_T3")
|
||||
_T4 = TypeVar("_T4")
|
||||
_T5 = TypeVar("_T5")
|
||||
_FT = TypeVar("_FT", bound=Future[Any])
|
||||
_FutureT = Union[Future[_T], Generator[Any, None, _T], Awaitable[_T]]
|
||||
_TaskYieldType = Optional[Future[object]]
|
||||
_FutureT = Future[_T] | Generator[Any, None, _T] | Awaitable[_T]
|
||||
_TaskYieldType = Future[object] | None
|
||||
|
||||
FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
|
||||
FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
|
||||
|
||||
@@ -7,7 +7,7 @@ from typing import Any, BinaryIO, Iterable, NoReturn, Union, overload
|
||||
# These are based in socket, maybe move them out into _typeshed.pyi or such
|
||||
_Address = Union[tuple[Any, ...], str]
|
||||
_RetAddress = Any
|
||||
_WriteBuffer = Union[bytearray, memoryview]
|
||||
_WriteBuffer = bytearray | memoryview
|
||||
_CMSG = tuple[int, int, bytes]
|
||||
|
||||
class TransportSocket:
|
||||
|
||||
Reference in New Issue
Block a user