.config/coc/extensions/node_modules/coc-pyright/node_modules/pyright/dist/typeshed-fallback/stdlib/asyncio/windows_events.pyi

   1 import socket
   2 import sys
   3 from _typeshed import WriteableBuffer
   4 from typing import IO, Any, Callable, ClassVar, NoReturn, Type
   5
   6 from . import events, futures, proactor_events, selector_events, streams, windows_utils
   7
   8 __all__ = [
   9     "SelectorEventLoop",
  10     "ProactorEventLoop",
  11     "IocpProactor",
  12     "DefaultEventLoopPolicy",
  13     "WindowsSelectorEventLoopPolicy",
  14     "WindowsProactorEventLoopPolicy",
  15 ]
  16
  17 NULL: int
  18 INFINITE: int
  19 ERROR_CONNECTION_REFUSED: int
  20 ERROR_CONNECTION_ABORTED: int
  21 CONNECT_PIPE_INIT_DELAY: float
  22 CONNECT_PIPE_MAX_DELAY: float
  23
  24 class PipeServer:
  25     def __init__(self, address: str) -> None: ...
  26     def __del__(self) -> None: ...
  27     def closed(self) -> bool: ...
  28     def close(self) -> None: ...
  29
  30 class _WindowsSelectorEventLoop(selector_events.BaseSelectorEventLoop): ...
  31
  32 class ProactorEventLoop(proactor_events.BaseProactorEventLoop):
  33     def __init__(self, proactor: IocpProactor | None = ...) -> None: ...
  34     async def create_pipe_connection(
  35         self, protocol_factory: Callable[[], streams.StreamReaderProtocol], address: str
  36     ) -> tuple[proactor_events._ProactorDuplexPipeTransport, streams.StreamReaderProtocol]: ...
  37     async def start_serving_pipe(
  38         self, protocol_factory: Callable[[], streams.StreamReaderProtocol], address: str
  39     ) -> list[PipeServer]: ...
  40
  41 class IocpProactor:
  42     def __init__(self, concurrency: int = ...) -> None: ...
  43     def __repr__(self) -> str: ...
  44     def __del__(self) -> None: ...
  45     def set_loop(self, loop: events.AbstractEventLoop) -> None: ...
  46     def select(self, timeout: int | None = ...) -> list[futures.Future[Any]]: ...
  47     def recv(self, conn: socket.socket, nbytes: int, flags: int = ...) -> futures.Future[bytes]: ...
  48     if sys.version_info >= (3, 7):
  49         def recv_into(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
  50     def send(self, conn: socket.socket, buf: WriteableBuffer, flags: int = ...) -> futures.Future[Any]: ...
  51     def accept(self, listener: socket.socket) -> futures.Future[Any]: ...
  52     def connect(self, conn: socket.socket, address: bytes) -> futures.Future[Any]: ...
  53     if sys.version_info >= (3, 7):
  54         def sendfile(self, sock: socket.socket, file: IO[bytes], offset: int, count: int) -> futures.Future[Any]: ...
  55     def accept_pipe(self, pipe: socket.socket) -> futures.Future[Any]: ...
  56     async def connect_pipe(self, address: bytes) -> windows_utils.PipeHandle: ...
  57     def wait_for_handle(self, handle: windows_utils.PipeHandle, timeout: int | None = ...) -> bool: ...
  58     def close(self) -> None: ...
  59
  60 SelectorEventLoop = _WindowsSelectorEventLoop
  61
  62 if sys.version_info >= (3, 7):
  63     class WindowsSelectorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
  64         _loop_factory: ClassVar[Type[SelectorEventLoop]]
  65         def get_child_watcher(self) -> NoReturn: ...
  66         def set_child_watcher(self, watcher: Any) -> NoReturn: ...
  67     class WindowsProactorEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
  68         _loop_factory: ClassVar[Type[ProactorEventLoop]]
  69         def get_child_watcher(self) -> NoReturn: ...
  70         def set_child_watcher(self, watcher: Any) -> NoReturn: ...
  71     DefaultEventLoopPolicy = WindowsSelectorEventLoopPolicy
  72 else:
  73     class _WindowsDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
  74         _loop_factory: ClassVar[Type[SelectorEventLoop]]
  75         def get_child_watcher(self) -> NoReturn: ...
  76         def set_child_watcher(self, watcher: Any) -> NoReturn: ...
  77     DefaultEventLoopPolicy = _WindowsDefaultEventLoopPolicy