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

Remove Python 3.6 branches from typeshed (#8269)

Browse Source
This commit is contained in:
Alex Waygood
2022-07-11 09:55:17 +01:00
committed by GitHub
parent 29c17ffb47
commit edc0ecd857
114 changed files with 1016 additions and 2642 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
stdlib/asyncio/__init__.pyi
View File

@@ -8,14 +8,12 @@ from .futures import *
from .locks import *
from .protocols import *
from .queues import *
from .runners import *
from .streams import *
from .subprocess import *
from .tasks import *
from .transports import *
if sys.version_info >= (3, 7):
from .runners import *
if sys.version_info >= (3, 8):
from .exceptions import *

153
stdlib/asyncio/base_events.pyi
View File

@@ -7,19 +7,15 @@ from asyncio.protocols import BaseProtocol
from asyncio.tasks import Task
from asyncio.transports import BaseTransport, ReadTransport, SubprocessTransport, WriteTransport
from collections.abc import Awaitable, Callable, Coroutine, Generator, Iterable, Sequence
from contextvars import Context
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, TypeVar, overload
from typing_extensions import Literal, TypeAlias
if sys.version_info >= (3, 7):
from contextvars import Context
if sys.version_info >= (3, 9):
__all__ = ("BaseEventLoop", "Server")
elif sys.version_info >= (3, 7):
__all__ = ("BaseEventLoop",)
else:
__all__ = ["BaseEventLoop"]
__all__ = ("BaseEventLoop",)
_T = TypeVar("_T")
_ProtocolT = TypeVar("_ProtocolT", bound=BaseProtocol)
@@ -40,7 +36,7 @@ class Server(AbstractServer):
ssl_handshake_timeout: float | None,
ssl_shutdown_timeout: float | None = ...,
) -> None: ...
elif sys.version_info >= (3, 7):
else:
def __init__(
self,
loop: AbstractEventLoop,
@@ -50,21 +46,18 @@ class Server(AbstractServer):
backlog: int,
ssl_handshake_timeout: float | None,
) -> None: ...
else:
def __init__(self, loop: AbstractEventLoop, sockets: list[socket]) -> None: ...
if sys.version_info >= (3, 7):
def get_loop(self) -> AbstractEventLoop: ...
def is_serving(self) -> bool: ...
async def start_serving(self) -> None: ...
async def serve_forever(self) -> None: ...
def get_loop(self) -> AbstractEventLoop: ...
def is_serving(self) -> bool: ...
async def start_serving(self) -> None: ...
async def serve_forever(self) -> None: ...
if sys.version_info >= (3, 8):
@property
def sockets(self) -> tuple[socket, ...]: ...
elif sys.version_info >= (3, 7):
else:
@property
def sockets(self) -> list[socket]: ...
else:
sockets: list[socket] | None
def close(self) -> None: ...
async def wait_closed(self) -> None: ...
@@ -81,19 +74,11 @@ class BaseEventLoop(AbstractEventLoop):
def close(self) -> None: ...
async def shutdown_asyncgens(self) -> None: ...
# Methods scheduling callbacks. All these return Handles.
if sys.version_info >= (3, 7):
def call_soon(self, callback: Callable[..., Any], *args: Any, context: Context | None = ...) -> Handle: ...
def call_later(
self, delay: float, callback: Callable[..., Any], *args: Any, context: Context | None = ...
) -> TimerHandle: ...
def call_at(
self, when: float, callback: Callable[..., Any], *args: Any, context: Context | None = ...
) -> TimerHandle: ...
else:
def call_soon(self, callback: Callable[..., Any], *args: Any) -> Handle: ...
def call_later(self, delay: float, callback: Callable[..., Any], *args: Any) -> TimerHandle: ...
def call_at(self, when: float, callback: Callable[..., Any], *args: Any) -> TimerHandle: ...
def call_soon(self, callback: Callable[..., Any], *args: Any, context: Context | None = ...) -> Handle: ...
def call_later(
self, delay: float, callback: Callable[..., Any], *args: Any, context: Context | None = ...
) -> TimerHandle: ...
def call_at(self, when: float, callback: Callable[..., Any], *args: Any, context: Context | None = ...) -> TimerHandle: ...
def time(self) -> float: ...
# Future methods
def create_future(self) -> Future[Any]: ...
@@ -110,11 +95,7 @@ class BaseEventLoop(AbstractEventLoop):
def set_task_factory(self, factory: _TaskFactory | None) -> None: ...
def get_task_factory(self) -> _TaskFactory | None: ...
# Methods for interacting with threads
if sys.version_info >= (3, 7):
def call_soon_threadsafe(self, callback: Callable[..., Any], *args: Any, context: Context | None = ...) -> Handle: ...
else:
def call_soon_threadsafe(self, callback: Callable[..., Any], *args: Any) -> Handle: ...
def call_soon_threadsafe(self, callback: Callable[..., Any], *args: Any, context: Context | None = ...) -> Handle: ...
def run_in_executor(self, executor: Any, func: Callable[..., _T], *args: Any) -> Future[_T]: ...
def set_default_executor(self, executor: Any) -> None: ...
# Network I/O methods returning Futures.
@@ -205,39 +186,6 @@ class BaseEventLoop(AbstractEventLoop):
happy_eyeballs_delay: float | None = ...,
interleave: int | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
elif sys.version_info >= (3, 7):
@overload
async def create_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
host: str = ...,
port: int = ...,
*,
ssl: _SSLContext = ...,
family: int = ...,
proto: int = ...,
flags: int = ...,
sock: None = ...,
local_addr: tuple[str, int] | None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
@overload
async def create_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
host: None = ...,
port: None = ...,
*,
ssl: _SSLContext = ...,
family: int = ...,
proto: int = ...,
flags: int = ...,
sock: socket,
local_addr: None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
else:
@overload
async def create_connection(
@@ -253,6 +201,7 @@ class BaseEventLoop(AbstractEventLoop):
sock: None = ...,
local_addr: tuple[str, int] | None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
@overload
async def create_connection(
@@ -268,6 +217,7 @@ class BaseEventLoop(AbstractEventLoop):
sock: socket,
local_addr: None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
if sys.version_info >= (3, 11):
@overload
@@ -326,7 +276,7 @@ class BaseEventLoop(AbstractEventLoop):
ssl_handshake_timeout: float | None = ...,
ssl_shutdown_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
elif sys.version_info >= (3, 7):
else:
@overload
async def create_server(
self,
@@ -379,47 +329,13 @@ class BaseEventLoop(AbstractEventLoop):
ssl: _SSLContext = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
else:
@overload
async def create_server(
self,
protocol_factory: _ProtocolFactory,
host: str | Sequence[str] | None = ...,
port: int = ...,
*,
family: int = ...,
flags: int = ...,
sock: None = ...,
backlog: int = ...,
ssl: _SSLContext = ...,
reuse_address: bool | None = ...,
reuse_port: bool | None = ...,
) -> Server: ...
@overload
async def create_server(
self,
protocol_factory: _ProtocolFactory,
host: None = ...,
port: None = ...,
*,
family: int = ...,
flags: int = ...,
sock: socket,
backlog: int = ...,
ssl: _SSLContext = ...,
reuse_address: bool | None = ...,
reuse_port: bool | None = ...,
) -> Server: ...
async def connect_accepted_socket(
self, protocol_factory: Callable[[], _ProtocolT], sock: socket, *, ssl: _SSLContext = ...
) -> tuple[BaseTransport, _ProtocolT]: ...
if sys.version_info >= (3, 7):
async def sock_sendfile(
self, sock: socket, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool | None = ...
) -> int: ...
async def sendfile(
self, transport: BaseTransport, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool = ...
) -> int: ...
async def sock_sendfile(
self, sock: socket, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool | None = ...
) -> int: ...
async def sendfile(
self, transport: BaseTransport, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool = ...
) -> int: ...
if sys.version_info >= (3, 11):
async def create_datagram_endpoint( # type: ignore[override]
self,
@@ -493,18 +409,11 @@ class BaseEventLoop(AbstractEventLoop):
def remove_writer(self, fd: FileDescriptorLike) -> bool: ...
# The sock_* methods (and probably some others) are not actually implemented on
# BaseEventLoop, only on subclasses. We list them here for now for convenience.
# Completion based I/O methods returning Futures prior to 3.7
if sys.version_info >= (3, 7):
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
async def sock_connect(self, sock: socket, address: _Address) -> None: ...
async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
else:
def sock_recv(self, sock: socket, nbytes: int) -> Future[bytes]: ...
def sock_sendall(self, sock: socket, data: bytes) -> Future[None]: ...
def sock_connect(self, sock: socket, address: _Address) -> Future[None]: ...
def sock_accept(self, sock: socket) -> Future[tuple[socket, _RetAddress]]: ...
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
async def sock_connect(self, sock: socket, address: _Address) -> None: ...
async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
if sys.version_info >= (3, 11):
async def sock_recvfrom(self, sock: socket, bufsize: int) -> bytes: ...
async def sock_recvfrom_into(self, sock: socket, buf: WriteableBuffer, nbytes: int = ...) -> int: ...

17
stdlib/asyncio/base_futures.pyi
View File

@@ -1,17 +1,11 @@
import sys
from collections.abc import Callable, Sequence
from contextvars import Context
from typing import Any
from typing_extensions import Literal
if sys.version_info >= (3, 7):
from contextvars import Context
from . import futures
if sys.version_info >= (3, 7):
__all__ = ()
else:
__all__: list[str] = []
__all__ = ()
# asyncio defines 'isfuture()' in base_futures.py and re-imports it in futures.py
# but it leads to circular import error in pytype tool.
@@ -22,10 +16,5 @@ _PENDING: Literal["PENDING"] # undocumented
_CANCELLED: Literal["CANCELLED"] # undocumented
_FINISHED: Literal["FINISHED"] # undocumented
if sys.version_info >= (3, 7):
def _format_callbacks(cb: Sequence[tuple[Callable[[futures.Future[Any]], None], Context]]) -> str: ... # undocumented
else:
def _format_callbacks(cb: Sequence[Callable[[futures.Future[Any]], None]]) -> str: ... # undocumented
def _format_callbacks(cb: Sequence[tuple[Callable[[futures.Future[Any]], None], Context]]) -> str: ... # undocumented
def _future_repr_info(future: futures.Future[Any]) -> list[str]: ... # undocumented

5
stdlib/asyncio/constants.pyi
View File

@@ -5,9 +5,8 @@ from typing_extensions import Literal
LOG_THRESHOLD_FOR_CONNLOST_WRITES: Literal[5]
ACCEPT_RETRY_DELAY: Literal[1]
DEBUG_STACK_DEPTH: Literal[10]
if sys.version_info >= (3, 7):
SSL_HANDSHAKE_TIMEOUT: float
SENDFILE_FALLBACK_READBUFFER_SIZE: Literal[262144]
SSL_HANDSHAKE_TIMEOUT: float
SENDFILE_FALLBACK_READBUFFER_SIZE: Literal[262144]
if sys.version_info >= (3, 11):
SSL_SHUTDOWN_TIMEOUT: float
FLOW_CONTROL_HIGH_WATER_SSL_READ: Literal[256]

4
stdlib/asyncio/coroutines.pyi
View File

@@ -5,10 +5,8 @@ from typing_extensions import TypeGuard
if sys.version_info >= (3, 11):
__all__ = ("iscoroutinefunction", "iscoroutine")
elif sys.version_info >= (3, 7):
__all__ = ("coroutine", "iscoroutinefunction", "iscoroutine")
else:
__all__ = ["coroutine", "iscoroutinefunction", "iscoroutine"]
__all__ = ("coroutine", "iscoroutinefunction", "iscoroutine")
if sys.version_info < (3, 11):
from collections.abc import Callable

230
stdlib/asyncio/events.pyi
View File

@@ -3,6 +3,7 @@ import sys
from _typeshed import FileDescriptorLike, Self, WriteableBuffer
from abc import ABCMeta, abstractmethod
from collections.abc import Awaitable, Callable, Coroutine, Generator, Sequence
from contextvars import Context
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, Protocol, TypeVar, overload
from typing_extensions import Literal, TypeAlias
@@ -14,9 +15,6 @@ from .tasks import Task
from .transports import BaseTransport, ReadTransport, SubprocessTransport, WriteTransport
from .unix_events import AbstractChildWatcher
if sys.version_info >= (3, 7):
from contextvars import Context
if sys.version_info >= (3, 8):
__all__ = (
"AbstractEventLoopPolicy",
@@ -36,7 +34,7 @@ if sys.version_info >= (3, 8):
"_get_running_loop",
)
elif sys.version_info >= (3, 7):
else:
__all__ = (
"AbstractEventLoopPolicy",
"AbstractEventLoop",
@@ -56,24 +54,6 @@ elif sys.version_info >= (3, 7):
"_get_running_loop",
)
else:
__all__ = [
"AbstractEventLoopPolicy",
"AbstractEventLoop",
"AbstractServer",
"Handle",
"TimerHandle",
"get_event_loop_policy",
"set_event_loop_policy",
"get_event_loop",
"set_event_loop",
"new_event_loop",
"get_child_watcher",
"set_child_watcher",
"_set_running_loop",
"_get_running_loop",
]
_T = TypeVar("_T")
_ProtocolT = TypeVar("_ProtocolT", bound=BaseProtocol)
_Context: TypeAlias = dict[str, Any]
@@ -89,35 +69,24 @@ class _TaskFactory(Protocol):
class Handle:
_cancelled: bool
_args: Sequence[Any]
if sys.version_info >= (3, 7):
def __init__(
self, callback: Callable[..., Any], args: Sequence[Any], loop: AbstractEventLoop, context: Context | None = ...
) -> None: ...
else:
def __init__(self, callback: Callable[..., Any], args: Sequence[Any], loop: AbstractEventLoop) -> None: ...
def __init__(
self, callback: Callable[..., Any], args: Sequence[Any], loop: AbstractEventLoop, context: Context | None = ...
) -> None: ...
def cancel(self) -> None: ...
def _run(self) -> None: ...
if sys.version_info >= (3, 7):
def cancelled(self) -> bool: ...
def cancelled(self) -> bool: ...
class TimerHandle(Handle):
if sys.version_info >= (3, 7):
def __init__(
self,
when: float,
callback: Callable[..., Any],
args: Sequence[Any],
loop: AbstractEventLoop,
context: Context | None = ...,
) -> None: ...
else:
def __init__(self, when: float, callback: Callable[..., Any], args: Sequence[Any], loop: AbstractEventLoop) -> None: ...
def __init__(
self,
when: float,
callback: Callable[..., Any],
args: Sequence[Any],
loop: AbstractEventLoop,
context: Context | None = ...,
) -> None: ...
def __hash__(self) -> int: ...
if sys.version_info >= (3, 7):
def when(self) -> float: ...
def when(self) -> float: ...
def __lt__(self, other: TimerHandle) -> bool: ...
def __le__(self, other: TimerHandle) -> bool: ...
def __gt__(self, other: TimerHandle) -> bool: ...
@@ -127,18 +96,16 @@ class TimerHandle(Handle):
class AbstractServer:
@abstractmethod
def close(self) -> None: ...
if sys.version_info >= (3, 7):
async def __aenter__(self: Self) -> Self: ...
async def __aexit__(self, *exc: object) -> None: ...
@abstractmethod
def get_loop(self) -> AbstractEventLoop: ...
@abstractmethod
def is_serving(self) -> bool: ...
@abstractmethod
async def start_serving(self) -> None: ...
@abstractmethod
async def serve_forever(self) -> None: ...
async def __aenter__(self: Self) -> Self: ...
async def __aexit__(self, *exc: object) -> None: ...
@abstractmethod
def get_loop(self) -> AbstractEventLoop: ...
@abstractmethod
def is_serving(self) -> bool: ...
@abstractmethod
async def start_serving(self) -> None: ...
@abstractmethod
async def serve_forever(self) -> None: ...
@abstractmethod
async def wait_closed(self) -> None: ...
@@ -317,41 +284,6 @@ class AbstractEventLoop:
happy_eyeballs_delay: float | None = ...,
interleave: int | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
elif sys.version_info >= (3, 7):
@overload
@abstractmethod
async def create_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
host: str = ...,
port: int = ...,
*,
ssl: _SSLContext = ...,
family: int = ...,
proto: int = ...,
flags: int = ...,
sock: None = ...,
local_addr: tuple[str, int] | None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
@overload
@abstractmethod
async def create_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
host: None = ...,
port: None = ...,
*,
ssl: _SSLContext = ...,
family: int = ...,
proto: int = ...,
flags: int = ...,
sock: socket,
local_addr: None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
else:
@overload
@abstractmethod
@@ -368,6 +300,7 @@ class AbstractEventLoop:
sock: None = ...,
local_addr: tuple[str, int] | None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
@overload
@abstractmethod
@@ -384,6 +317,7 @@ class AbstractEventLoop:
sock: socket,
local_addr: None = ...,
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
if sys.version_info >= (3, 11):
@overload
@@ -448,7 +382,7 @@ class AbstractEventLoop:
ssl_shutdown_timeout: float | None = ...,
start_serving: bool = ...,
) -> Server: ...
elif sys.version_info >= (3, 7):
else:
@overload
@abstractmethod
async def create_server(
@@ -507,48 +441,6 @@ class AbstractEventLoop:
ssl_handshake_timeout: float | None = ...,
start_serving: bool = ...,
) -> Server: ...
else:
@overload
@abstractmethod
async def create_server(
self,
protocol_factory: _ProtocolFactory,
host: str | Sequence[str] | None = ...,
port: int = ...,
*,
family: int = ...,
flags: int = ...,
sock: None = ...,
backlog: int = ...,
ssl: _SSLContext = ...,
reuse_address: bool | None = ...,
reuse_port: bool | None = ...,
) -> Server: ...
@overload
@abstractmethod
async def create_server(
self,
protocol_factory: _ProtocolFactory,
host: None = ...,
port: None = ...,
*,
family: int = ...,
flags: int = ...,
sock: socket,
backlog: int = ...,
ssl: _SSLContext = ...,
reuse_address: bool | None = ...,
reuse_port: bool | None = ...,
) -> Server: ...
async def create_unix_server(
self,
protocol_factory: _ProtocolFactory,
path: str,
*,
sock: socket | None = ...,
backlog: int = ...,
ssl: _SSLContext = ...,
) -> Server: ...
if sys.version_info >= (3, 11):
async def connect_accepted_socket(
self,
@@ -580,7 +472,7 @@ class AbstractEventLoop:
ssl_handshake_timeout: float | None = ...,
ssl_shutdown_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
elif sys.version_info >= (3, 7):
else:
async def create_unix_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
@@ -591,26 +483,15 @@ class AbstractEventLoop:
server_hostname: str | None = ...,
ssl_handshake_timeout: float | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
else:
async def create_unix_connection(
self,
protocol_factory: Callable[[], _ProtocolT],
path: str,
*,
ssl: _SSLContext = ...,
sock: socket | None = ...,
server_hostname: str | None = ...,
) -> tuple[BaseTransport, _ProtocolT]: ...
if sys.version_info >= (3, 7):
@abstractmethod
async def sock_sendfile(
self, sock: socket, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool | None = ...
) -> int: ...
@abstractmethod
async def sendfile(
self, transport: BaseTransport, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool = ...
) -> int: ...
@abstractmethod
async def sock_sendfile(
self, sock: socket, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool | None = ...
) -> int: ...
@abstractmethod
async def sendfile(
self, transport: BaseTransport, file: IO[bytes], offset: int = ..., count: int | None = ..., *, fallback: bool = ...
) -> int: ...
@abstractmethod
async def create_datagram_endpoint(
self,
@@ -677,26 +558,16 @@ class AbstractEventLoop:
@abstractmethod
def remove_writer(self, fd: FileDescriptorLike) -> bool: ...
# Completion based I/O methods returning Futures prior to 3.7
if sys.version_info >= (3, 7):
@abstractmethod
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
@abstractmethod
async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
@abstractmethod
async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
@abstractmethod
async def sock_connect(self, sock: socket, address: _Address) -> None: ...
@abstractmethod
async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
else:
@abstractmethod
def sock_recv(self, sock: socket, nbytes: int) -> Future[bytes]: ...
@abstractmethod
def sock_sendall(self, sock: socket, data: bytes) -> Future[None]: ...
@abstractmethod
def sock_connect(self, sock: socket, address: _Address) -> Future[None]: ...
@abstractmethod
def sock_accept(self, sock: socket) -> Future[tuple[socket, _RetAddress]]: ...
@abstractmethod
async def sock_recv(self, sock: socket, nbytes: int) -> bytes: ...
@abstractmethod
async def sock_recv_into(self, sock: socket, buf: WriteableBuffer) -> int: ...
@abstractmethod
async def sock_sendall(self, sock: socket, data: bytes) -> None: ...
@abstractmethod
async def sock_connect(self, sock: socket, address: _Address) -> None: ...
@abstractmethod
async def sock_accept(self, sock: socket) -> tuple[socket, _RetAddress]: ...
if sys.version_info >= (3, 11):
@abstractmethod
async def sock_recvfrom(self, sock: socket, bufsize: int) -> bytes: ...
@@ -755,8 +626,7 @@ def get_child_watcher() -> AbstractChildWatcher: ...
def set_child_watcher(watcher: AbstractChildWatcher) -> None: ...
def _set_running_loop(__loop: AbstractEventLoop | None) -> None: ...
def _get_running_loop() -> AbstractEventLoop: ...
def get_running_loop() -> AbstractEventLoop: ...
if sys.version_info >= (3, 7):
def get_running_loop() -> AbstractEventLoop: ...
if sys.version_info < (3, 8):
class SendfileNotAvailableError(RuntimeError): ...
if sys.version_info < (3, 8):
class SendfileNotAvailableError(RuntimeError): ...

29
stdlib/asyncio/futures.pyi
View File

@@ -12,18 +12,15 @@ if sys.version_info < (3, 8):
class InvalidStateError(Error): ...
if sys.version_info >= (3, 7):
from contextvars import Context
from contextvars import Context
if sys.version_info >= (3, 9):
from types import GenericAlias
if sys.version_info >= (3, 8):
__all__ = ("Future", "wrap_future", "isfuture")
elif sys.version_info >= (3, 7):
__all__ = ("CancelledError", "TimeoutError", "InvalidStateError", "Future", "wrap_future", "isfuture")
else:
__all__ = ["CancelledError", "TimeoutError", "InvalidStateError", "Future", "wrap_future", "isfuture"]
__all__ = ("CancelledError", "TimeoutError", "InvalidStateError", "Future", "wrap_future", "isfuture")
_T = TypeVar("_T")
@@ -32,15 +29,6 @@ _T = TypeVar("_T")
# That's why the import order is reversed.
def isfuture(obj: object) -> TypeGuard[Future[Any]]: ...
if sys.version_info < (3, 7):
class _TracebackLogger:
exc: BaseException
tb: list[str]
def __init__(self, exc: Any, loop: AbstractEventLoop) -> None: ...
def activate(self) -> None: ...
def clear(self) -> None: ...
def __del__(self) -> None: ...
class Future(Awaitable[_T], Iterable[_T]):
_state: str
@property
@@ -53,15 +41,10 @@ class Future(Awaitable[_T], Iterable[_T]):
_asyncio_future_blocking: bool # is a part of duck-typing contract for `Future`
def __init__(self, *, loop: AbstractEventLoop | None = ...) -> None: ...
def __del__(self) -> None: ...
if sys.version_info >= (3, 7):
def get_loop(self) -> AbstractEventLoop: ...
@property
def _callbacks(self: Self) -> list[tuple[Callable[[Self], Any], Context]]: ...
def add_done_callback(self: Self, __fn: Callable[[Self], Any], *, context: Context | None = ...) -> None: ...
else:
@property
def _callbacks(self: Self) -> list[Callable[[Self], Any]]: ...
def add_done_callback(self: Self, __fn: Callable[[Self], Any]) -> None: ...
def get_loop(self) -> AbstractEventLoop: ...
@property
def _callbacks(self: Self) -> list[tuple[Callable[[Self], Any], Context]]: ...
def add_done_callback(self: Self, __fn: Callable[[Self], Any], *, context: Context | None = ...) -> None: ...
if sys.version_info >= (3, 9):
def cancel(self, msg: Any | None = ...) -> bool: ...
else:

4
stdlib/asyncio/locks.pyi
View File

@@ -15,10 +15,8 @@ if sys.version_info >= (3, 11):
if sys.version_info >= (3, 11):
__all__ = ("Lock", "Event", "Condition", "Semaphore", "BoundedSemaphore", "Barrier")
elif sys.version_info >= (3, 7):
__all__ = ("Lock", "Event", "Condition", "Semaphore", "BoundedSemaphore")
else:
__all__ = ["Lock", "Event", "Condition", "Semaphore", "BoundedSemaphore"]
__all__ = ("Lock", "Event", "Condition", "Semaphore", "BoundedSemaphore")
_T = TypeVar("_T")

5
stdlib/asyncio/proactor_events.pyi
View File

@@ -6,10 +6,7 @@ from typing_extensions import Literal
from . import base_events, constants, events, futures, streams, transports
if sys.version_info >= (3, 7):
__all__ = ("BaseProactorEventLoop",)
else:
__all__ = ["BaseProactorEventLoop"]
__all__ = ("BaseProactorEventLoop",)
if sys.version_info >= (3, 8):
class _WarnCallbackProtocol(Protocol):

15
stdlib/asyncio/protocols.pyi
View File

@@ -1,12 +1,8 @@
import sys
from _typeshed import ReadableBuffer
from asyncio import transports
from typing import Any
if sys.version_info >= (3, 7):
__all__ = ("BaseProtocol", "Protocol", "DatagramProtocol", "SubprocessProtocol", "BufferedProtocol")
else:
__all__ = ["BaseProtocol", "Protocol", "DatagramProtocol", "SubprocessProtocol"]
__all__ = ("BaseProtocol", "Protocol", "DatagramProtocol", "SubprocessProtocol", "BufferedProtocol")
class BaseProtocol:
def connection_made(self, transport: transports.BaseTransport) -> None: ...
@@ -18,11 +14,10 @@ class Protocol(BaseProtocol):
def data_received(self, data: bytes) -> None: ...
def eof_received(self) -> bool | None: ...
if sys.version_info >= (3, 7):
class BufferedProtocol(BaseProtocol):
def get_buffer(self, sizehint: int) -> ReadableBuffer: ...
def buffer_updated(self, nbytes: int) -> None: ...
def eof_received(self) -> bool | None: ...
class BufferedProtocol(BaseProtocol):
def get_buffer(self, sizehint: int) -> ReadableBuffer: ...
def buffer_updated(self, nbytes: int) -> None: ...
def eof_received(self) -> bool | None: ...
class DatagramProtocol(BaseProtocol):
def connection_made(self, transport: transports.DatagramTransport) -> None: ... # type: ignore[override]

5
stdlib/asyncio/queues.pyi
View File

@@ -5,10 +5,7 @@ from typing import Any, Generic, TypeVar
if sys.version_info >= (3, 9):
from types import GenericAlias
if sys.version_info >= (3, 7):
__all__ = ("Queue", "PriorityQueue", "LifoQueue", "QueueFull", "QueueEmpty")
else:
__all__ = ["Queue", "PriorityQueue", "LifoQueue", "QueueFull", "QueueEmpty"]
__all__ = ("Queue", "PriorityQueue", "LifoQueue", "QueueFull", "QueueEmpty")
class QueueEmpty(Exception): ...
class QueueFull(Exception): ...

6
stdlib/asyncio/selector_events.pyi
View File

@@ -1,12 +1,8 @@
import selectors
import sys
from . import base_events
if sys.version_info >= (3, 7):
__all__ = ("BaseSelectorEventLoop",)
else:
__all__ = ["BaseSelectorEventLoop"]
__all__ = ("BaseSelectorEventLoop",)
class BaseSelectorEventLoop(base_events.BaseEventLoop):
def __init__(self, selector: selectors.BaseSelector | None = ...) -> None: ...

30
stdlib/asyncio/sslproto.pyi
View File

@@ -76,17 +76,13 @@ class _SSLProtocolTransport(transports._FlowControlMixin, transports.Transport):
def get_protocol(self) -> protocols.BaseProtocol: ...
def is_closing(self) -> bool: ...
def close(self) -> None: ...
if sys.version_info >= (3, 7):
def is_reading(self) -> bool: ...
def is_reading(self) -> bool: ...
def pause_reading(self) -> None: ...
def resume_reading(self) -> None: ...
def set_write_buffer_limits(self, high: int | None = ..., low: int | None = ...) -> None: ...
def get_write_buffer_size(self) -> int: ...
if sys.version_info >= (3, 7):
@property
def _protocol_paused(self) -> bool: ...
@property
def _protocol_paused(self) -> bool: ...
def write(self, data: bytes) -> None: ...
def can_write_eof(self) -> Literal[False]: ...
def abort(self) -> None: ...
@@ -138,7 +134,7 @@ class SSLProtocol(_SSLProtocolBase):
ssl_handshake_timeout: int | None = ...,
ssl_shutdown_timeout: float | None = ...,
) -> None: ...
elif sys.version_info >= (3, 7):
else:
def __init__(
self,
loop: events.AbstractEventLoop,
@@ -150,20 +146,8 @@ class SSLProtocol(_SSLProtocolBase):
call_connection_made: bool = ...,
ssl_handshake_timeout: int | None = ...,
) -> None: ...
else:
def __init__(
self,
loop: events.AbstractEventLoop,
app_protocol: protocols.BaseProtocol,
sslcontext: ssl.SSLContext,
waiter: futures.Future[Any],
server_side: bool = ...,
server_hostname: str | None = ...,
call_connection_made: bool = ...,
) -> None: ...
if sys.version_info >= (3, 7):
def _set_app_protocol(self, app_protocol: protocols.BaseProtocol) -> None: ...
def _set_app_protocol(self, app_protocol: protocols.BaseProtocol) -> None: ...
def _wakeup_waiter(self, exc: BaseException | None = ...) -> None: ...
def connection_made(self, transport: transports.BaseTransport) -> None: ...
def connection_lost(self, exc: BaseException | None) -> None: ...
@@ -178,9 +162,7 @@ class SSLProtocol(_SSLProtocolBase):
def _write_appdata(self, data: bytes) -> None: ...
def _start_handshake(self) -> None: ...
if sys.version_info >= (3, 7):
def _check_handshake_timeout(self) -> None: ...
def _check_handshake_timeout(self) -> None: ...
def _on_handshake_complete(self, handshake_exc: BaseException | None) -> None: ...
def _fatal_error(self, exc: BaseException, message: str = ...) -> None: ...
def _abort(self) -> None: ...

44
stdlib/asyncio/streams.pyi
View File

@@ -11,7 +11,7 @@ from .base_events import Server
if sys.platform == "win32":
if sys.version_info >= (3, 8):
__all__ = ("StreamReader", "StreamWriter", "StreamReaderProtocol", "open_connection", "start_server")
elif sys.version_info >= (3, 7):
else:
__all__ = (
"StreamReader",
"StreamWriter",
@@ -21,16 +21,6 @@ if sys.platform == "win32":
"IncompleteReadError",
"LimitOverrunError",
)
else:
__all__ = [
"StreamReader",
"StreamWriter",
"StreamReaderProtocol",
"open_connection",
"start_server",
"IncompleteReadError",
"LimitOverrunError",
]
else:
if sys.version_info >= (3, 8):
__all__ = (
@@ -42,7 +32,7 @@ else:
"open_unix_connection",
"start_unix_server",
)
elif sys.version_info >= (3, 7):
else:
__all__ = (
"StreamReader",
"StreamWriter",
@@ -54,18 +44,6 @@ else:
"open_unix_connection",
"start_unix_server",
)
else:
__all__ = [
"StreamReader",
"StreamWriter",
"StreamReaderProtocol",
"open_connection",
"start_server",
"IncompleteReadError",
"LimitOverrunError",
"open_unix_connection",
"start_unix_server",
]
_ClientConnectedCallback: TypeAlias = Callable[[StreamReader, StreamWriter], Awaitable[None] | None]
@@ -120,24 +98,20 @@ else:
) -> Server: ...
if sys.platform != "win32":
if sys.version_info >= (3, 7):
_PathType: TypeAlias = StrPath
else:
_PathType: TypeAlias = str
if sys.version_info >= (3, 10):
async def open_unix_connection(
path: _PathType | None = ..., *, limit: int = ..., **kwds: Any
path: StrPath | None = ..., *, limit: int = ..., **kwds: Any
) -> tuple[StreamReader, StreamWriter]: ...
async def start_unix_server(
client_connected_cb: _ClientConnectedCallback, path: _PathType | None = ..., *, limit: int = ..., **kwds: Any
client_connected_cb: _ClientConnectedCallback, path: StrPath | None = ..., *, limit: int = ..., **kwds: Any
) -> Server: ...
else:
async def open_unix_connection(
path: _PathType | None = ..., *, loop: events.AbstractEventLoop | None = ..., limit: int = ..., **kwds: Any
path: StrPath | None = ..., *, loop: events.AbstractEventLoop | None = ..., limit: int = ..., **kwds: Any
) -> tuple[StreamReader, StreamWriter]: ...
async def start_unix_server(
client_connected_cb: _ClientConnectedCallback,
path: _PathType | None = ...,
path: StrPath | None = ...,
*,
loop: events.AbstractEventLoop | None = ...,
limit: int = ...,
@@ -174,10 +148,8 @@ class StreamWriter:
def write_eof(self) -> None: ...
def can_write_eof(self) -> bool: ...
def close(self) -> None: ...
if sys.version_info >= (3, 7):
def is_closing(self) -> bool: ...
async def wait_closed(self) -> None: ...
def is_closing(self) -> bool: ...
async def wait_closed(self) -> None: ...
def get_extra_info(self, name: str, default: Any = ...) -> Any: ...
async def drain(self) -> None: ...
if sys.version_info >= (3, 11):

5
stdlib/asyncio/subprocess.pyi
View File

@@ -6,10 +6,7 @@ from collections.abc import Callable
from typing import IO, Any
from typing_extensions import Literal, TypeAlias
if sys.version_info >= (3, 7):
__all__ = ("create_subprocess_exec", "create_subprocess_shell")
else:
__all__ = ["create_subprocess_exec", "create_subprocess_shell"]
__all__ = ("create_subprocess_exec", "create_subprocess_shell")
if sys.version_info >= (3, 8):
_ExecArg: TypeAlias = StrOrBytesPath

95
stdlib/asyncio/tasks.pyi
View File

@@ -13,43 +13,27 @@ if sys.version_info >= (3, 9):
if sys.version_info >= (3, 11):
from contextvars import Context
if sys.version_info >= (3, 7):
__all__ = (
"Task",
"create_task",
"FIRST_COMPLETED",
"FIRST_EXCEPTION",
"ALL_COMPLETED",
"wait",
"wait_for",
"as_completed",
"sleep",
"gather",
"shield",
"ensure_future",
"run_coroutine_threadsafe",
"current_task",
"all_tasks",
"_register_task",
"_unregister_task",
"_enter_task",
"_leave_task",
)
else:
__all__ = [
"Task",
"FIRST_COMPLETED",
"FIRST_EXCEPTION",
"ALL_COMPLETED",
"wait",
"wait_for",
"as_completed",
"sleep",
"gather",
"shield",
"ensure_future",
"run_coroutine_threadsafe",
]
__all__ = (
"Task",
"create_task",
"FIRST_COMPLETED",
"FIRST_EXCEPTION",
"ALL_COMPLETED",
"wait",
"wait_for",
"as_completed",
"sleep",
"gather",
"shield",
"ensure_future",
"run_coroutine_threadsafe",
"current_task",
"all_tasks",
"_register_task",
"_unregister_task",
"_enter_task",
"_leave_task",
)
_T = TypeVar("_T")
_T1 = TypeVar("_T1")
@@ -78,9 +62,6 @@ def ensure_future(coro_or_future: _FT, *, loop: AbstractEventLoop | None = ...)
@overload
def ensure_future(coro_or_future: Awaitable[_T], *, loop: AbstractEventLoop | None = ...) -> Task[_T]: ...
# Prior to Python 3.7 'async' was an alias for 'ensure_future'.
# It became a keyword in 3.7.
# `gather()` actually returns a list with length equal to the number
# of tasks passed; however, Tuple is used similar to the annotation for
# zip() because typing does not support variadic type variables. See
@@ -334,24 +315,24 @@ class Task(Future[_T], Generic[_T]):
def current_task(cls, loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...
@classmethod
def all_tasks(cls, loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
if sys.version_info < (3, 7):
def _wakeup(self, fut: Future[Any]) -> None: ...
if sys.version_info >= (3, 9):
def __class_getitem__(cls, item: Any) -> GenericAlias: ...
if sys.version_info >= (3, 7):
def all_tasks(loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
if sys.version_info >= (3, 11):
def create_task(
coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = ..., context: Context | None = ...
) -> Task[_T]: ...
elif sys.version_info >= (3, 8):
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = ...) -> Task[_T]: ...
else:
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T]) -> Task[_T]: ...
def all_tasks(loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
def current_task(loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...
def _enter_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...
def _leave_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...
def _register_task(task: Task[Any]) -> None: ...
def _unregister_task(task: Task[Any]) -> None: ...
if sys.version_info >= (3, 11):
def create_task(
coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = ..., context: Context | None = ...
) -> Task[_T]: ...
elif sys.version_info >= (3, 8):
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = ...) -> Task[_T]: ...
else:
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T]) -> Task[_T]: ...
def current_task(loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...
def _enter_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...
def _leave_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...
def _register_task(task: Task[Any]) -> None: ...
def _unregister_task(task: Task[Any]) -> None: ...

10
stdlib/asyncio/transports.pyi
View File

@@ -1,14 +1,10 @@
import sys
from asyncio.events import AbstractEventLoop
from asyncio.protocols import BaseProtocol
from collections.abc import Mapping
from socket import _Address
from typing import Any
if sys.version_info >= (3, 7):
__all__ = ("BaseTransport", "ReadTransport", "WriteTransport", "Transport", "DatagramTransport", "SubprocessTransport")
else:
__all__ = ["BaseTransport", "ReadTransport", "WriteTransport", "Transport", "DatagramTransport", "SubprocessTransport"]
__all__ = ("BaseTransport", "ReadTransport", "WriteTransport", "Transport", "DatagramTransport", "SubprocessTransport")
class BaseTransport:
def __init__(self, extra: Mapping[Any, Any] | None = ...) -> None: ...
@@ -19,9 +15,7 @@ class BaseTransport:
def get_protocol(self) -> BaseProtocol: ...
class ReadTransport(BaseTransport):
if sys.version_info >= (3, 7):
def is_reading(self) -> bool: ...
def is_reading(self) -> bool: ...
def pause_reading(self) -> None: ...
def resume_reading(self) -> None: ...

20
stdlib/asyncio/unix_events.pyi
View File

@@ -3,12 +3,10 @@ import types
from _typeshed import Self
from abc import ABCMeta, abstractmethod
from collections.abc import Callable
from socket import socket
from typing import Any
from typing_extensions import Literal
from .base_events import Server
from .events import AbstractEventLoop, BaseDefaultEventLoopPolicy, _ProtocolFactory, _SSLContext
from .events import AbstractEventLoop, BaseDefaultEventLoopPolicy
from .selector_events import BaseSelectorEventLoop
# This is also technically not available on Win,
@@ -53,10 +51,8 @@ if sys.platform != "win32":
"ThreadedChildWatcher",
"DefaultEventLoopPolicy",
)
elif sys.version_info >= (3, 7):
__all__ = ("SelectorEventLoop", "AbstractChildWatcher", "SafeChildWatcher", "FastChildWatcher", "DefaultEventLoopPolicy")
else:
__all__ = ["SelectorEventLoop", "AbstractChildWatcher", "SafeChildWatcher", "FastChildWatcher", "DefaultEventLoopPolicy"]
__all__ = ("SelectorEventLoop", "AbstractChildWatcher", "SafeChildWatcher", "FastChildWatcher", "DefaultEventLoopPolicy")
# Doesn't actually have ABCMeta metaclass at runtime, but mypy complains if we don't have it in the stub.
# See discussion in #7412
@@ -80,17 +76,7 @@ if sys.platform != "win32":
def add_child_handler(self, pid: int, callback: Callable[..., Any], *args: Any) -> None: ...
def remove_child_handler(self, pid: int) -> bool: ...
class _UnixSelectorEventLoop(BaseSelectorEventLoop):
if sys.version_info < (3, 7):
async def create_unix_server(
self,
protocol_factory: _ProtocolFactory,
path: str | None = ...,
*,
sock: socket | None = ...,
backlog: int = ...,
ssl: _SSLContext = ...,
) -> Server: ...
class _UnixSelectorEventLoop(BaseSelectorEventLoop): ...
class _UnixDefaultEventLoopPolicy(BaseDefaultEventLoopPolicy):
def get_child_watcher(self) -> AbstractChildWatcher: ...

52
stdlib/asyncio/windows_events.pyi
View File

@@ -8,17 +8,14 @@ from typing_extensions import Literal
from . import events, futures, proactor_events, selector_events, streams, windows_utils
if sys.platform == "win32":
if sys.version_info >= (3, 7):
__all__ = (
"SelectorEventLoop",
"ProactorEventLoop",
"IocpProactor",
"DefaultEventLoopPolicy",
"WindowsSelectorEventLoopPolicy",
"WindowsProactorEventLoopPolicy",
)
else:
__all__ = ["SelectorEventLoop", "ProactorEventLoop", "IocpProactor", "DefaultEventLoopPolicy"]
__all__ = (
"SelectorEventLoop",
"ProactorEventLoop",
"IocpProactor",
"DefaultEventLoopPolicy",
"WindowsSelectorEventLoopPolicy",
"WindowsProactorEventLoopPolicy",
)
NULL: Literal[0]
INFINITE: Literal[0xFFFFFFFF]
@@ -50,35 +47,24 @@ if sys.platform == "win32":
def set_loop(self, loop: events.AbstractEventLoop) -> None: ...
def select(self, timeout: int | None = ...) -> list[futures.Future[Any]]: ...
def recv(self, conn: socket.socket, nbytes: int, flags: int = ...) -> futures.Future[bytes]: ...
if sys.version_info >= (3, 7):
def recv_into(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
def recv_into(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
def send(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
def accept(self, listener: socket.socket) -> futures.Future[Any]: ...
def connect(self, conn: socket.socket, address: bytes) -> futures.Future[Any]: ...
if sys.version_info >= (3, 7):
def sendfile(self, sock: socket.socket, file: IO[bytes], offset: int, count: int) -> futures.Future[Any]: ...
def sendfile(self, sock: socket.socket, file: IO[bytes], offset: int, count: int) -> futures.Future[Any]: ...
def accept_pipe(self, pipe: socket.socket) -> futures.Future[Any]: ...
async def connect_pipe(self, address: bytes) -> windows_utils.PipeHandle: ...
def wait_for_handle(self, handle: windows_utils.PipeHandle, timeout: int | None = ...) -> bool: ...
def close(self) -> None: ...
SelectorEventLoop = _WindowsSelectorEventLoop
if sys.version_info >= (3, 7):
class WindowsSelectorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
_loop_factory: ClassVar[type[SelectorEventLoop]]
def get_child_watcher(self) -> NoReturn: ...
def set_child_watcher(self, watcher: Any) -> NoReturn: ...
class WindowsSelectorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
_loop_factory: ClassVar[type[SelectorEventLoop]]
def get_child_watcher(self) -> NoReturn: ...
def set_child_watcher(self, watcher: Any) -> NoReturn: ...
class WindowsProactorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
_loop_factory: ClassVar[type[ProactorEventLoop]]
def get_child_watcher(self) -> NoReturn: ...
def set_child_watcher(self, watcher: Any) -> NoReturn: ...
DefaultEventLoopPolicy = WindowsSelectorEventLoopPolicy
else:
class _WindowsDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
_loop_factory: ClassVar[type[SelectorEventLoop]]
def get_child_watcher(self) -> NoReturn: ...
def set_child_watcher(self, watcher: Any) -> NoReturn: ...
DefaultEventLoopPolicy = _WindowsDefaultEventLoopPolicy
class WindowsProactorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
_loop_factory: ClassVar[type[ProactorEventLoop]]
def get_child_watcher(self) -> NoReturn: ...
def set_child_watcher(self, watcher: Any) -> NoReturn: ...
DefaultEventLoopPolicy = WindowsSelectorEventLoopPolicy

8
stdlib/asyncio/windows_utils.pyi
View File

@@ -7,13 +7,7 @@ from typing import Any, AnyStr, Protocol
from typing_extensions import Literal
if sys.platform == "win32":
if sys.version_info >= (3, 7):
__all__ = ("pipe", "Popen", "PIPE", "PipeHandle")
else:
__all__ = ["socketpair", "pipe", "Popen", "PIPE", "PipeHandle"]
import socket
socketpair = socket.socketpair
__all__ = ("pipe", "Popen", "PIPE", "PipeHandle")
class _WarnFunction(Protocol):
def __call__(