646 lines
25 KiB
Python
646 lines
25 KiB
Python
import ssl
|
|
import sys
|
|
from _asyncio import (
|
|
_get_running_loop as _get_running_loop,
|
|
_set_running_loop as _set_running_loop,
|
|
get_event_loop as get_event_loop,
|
|
get_running_loop as get_running_loop,
|
|
)
|
|
from _typeshed import FileDescriptorLike, ReadableBuffer, StrPath, Unused, WriteableBuffer
|
|
from abc import ABCMeta, abstractmethod
|
|
from collections.abc import Callable, Sequence
|
|
from contextvars import Context # type: ignore
|
|
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
|
|
from typing import IO, Any, Literal, Protocol, TypeVar, overload
|
|
from typing_extensions import Self, TypeAlias, TypeVarTuple, Unpack, deprecated
|
|
|
|
from . import _AwaitableLike, _CoroutineLike
|
|
from .base_events import Server
|
|
from .futures import Future
|
|
from .protocols import BaseProtocol
|
|
from .tasks import Task
|
|
from .transports import BaseTransport, DatagramTransport, ReadTransport, SubprocessTransport, Transport, WriteTransport
|
|
from .unix_events import AbstractChildWatcher
|
|
|
|
if sys.version_info >= (3, 14):
|
|
__all__ = (
|
|
"AbstractEventLoopPolicy",
|
|
"AbstractEventLoop",
|
|
"AbstractServer",
|
|
"Handle",
|
|
"TimerHandle",
|
|
"get_event_loop_policy",
|
|
"set_event_loop_policy",
|
|
"get_event_loop",
|
|
"set_event_loop",
|
|
"new_event_loop",
|
|
"_set_running_loop",
|
|
"get_running_loop",
|
|
"_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",
|
|
"_get_running_loop",
|
|
)
|
|
|
|
_T = TypeVar("_T")
|
|
_Ts = TypeVarTuple("_Ts")
|
|
_ProtocolT = TypeVar("_ProtocolT", bound=BaseProtocol)
|
|
_Context: TypeAlias = dict[str, Any]
|
|
_ExceptionHandler: TypeAlias = Callable[[AbstractEventLoop, _Context], object]
|
|
_ProtocolFactory: TypeAlias = Callable[[], BaseProtocol]
|
|
_SSLContext: TypeAlias = bool | None | ssl.SSLContext
|
|
|
|
class _TaskFactory(Protocol):
|
|
def __call__(self, loop: AbstractEventLoop, factory: _CoroutineLike[_T], /) -> Future[_T]: ...
|
|
|
|
class Handle:
|
|
_cancelled: bool
|
|
_args: Sequence[Any]
|
|
def __init__(
|
|
self, callback: Callable[..., object], args: Sequence[Any], loop: AbstractEventLoop, context: Context | None = None # type: ignore
|
|
) -> None: ...
|
|
def cancel(self) -> None: ...
|
|
def _run(self) -> None: ...
|
|
def cancelled(self) -> bool: ...
|
|
if sys.version_info >= (3, 12):
|
|
def get_context(self) -> Context: ...
|
|
|
|
class TimerHandle(Handle):
|
|
def __init__(
|
|
self,
|
|
when: float,
|
|
callback: Callable[..., object],
|
|
args: Sequence[Any],
|
|
loop: AbstractEventLoop,
|
|
context: Context | None = None, # type: ignore
|
|
) -> None: ...
|
|
def __hash__(self) -> int: ...
|
|
def when(self) -> float: ...
|
|
def __lt__(self, other: TimerHandle) -> bool: ...
|
|
def __le__(self, other: TimerHandle) -> bool: ...
|
|
def __gt__(self, other: TimerHandle) -> bool: ...
|
|
def __ge__(self, other: TimerHandle) -> bool: ...
|
|
def __eq__(self, other: object) -> bool: ...
|
|
|
|
class AbstractServer:
|
|
@abstractmethod
|
|
def close(self) -> None: ...
|
|
if sys.version_info >= (3, 13):
|
|
@abstractmethod
|
|
def close_clients(self) -> None: ...
|
|
@abstractmethod
|
|
def abort_clients(self) -> None: ...
|
|
|
|
async def __aenter__(self) -> Self: ...
|
|
async def __aexit__(self, *exc: Unused) -> 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: ...
|
|
|
|
class AbstractEventLoop:
|
|
slow_callback_duration: float
|
|
@abstractmethod
|
|
def run_forever(self) -> None: ...
|
|
@abstractmethod
|
|
def run_until_complete(self, future: _AwaitableLike[_T]) -> _T: ...
|
|
@abstractmethod
|
|
def stop(self) -> None: ...
|
|
@abstractmethod
|
|
def is_running(self) -> bool: ...
|
|
@abstractmethod
|
|
def is_closed(self) -> bool: ...
|
|
@abstractmethod
|
|
def close(self) -> None: ...
|
|
@abstractmethod
|
|
async def shutdown_asyncgens(self) -> None: ...
|
|
# Methods scheduling callbacks. All these return Handles.
|
|
if sys.version_info >= (3, 9): # "context" added in 3.9.10/3.10.2
|
|
@abstractmethod
|
|
def call_soon(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], context: Context | None = None) -> Handle: ... # type: ignore
|
|
@abstractmethod
|
|
def call_later(
|
|
self, delay: float, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], context: Context | None = None # type: ignore
|
|
) -> TimerHandle: ...
|
|
@abstractmethod
|
|
def call_at(
|
|
self, when: float, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], context: Context | None = None # type: ignore
|
|
) -> TimerHandle: ...
|
|
else:
|
|
@abstractmethod
|
|
def call_soon(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> Handle: ...
|
|
@abstractmethod
|
|
def call_later(self, delay: float, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> TimerHandle: ...
|
|
@abstractmethod
|
|
def call_at(self, when: float, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> TimerHandle: ...
|
|
|
|
@abstractmethod
|
|
def time(self) -> float: ...
|
|
# Future methods
|
|
@abstractmethod
|
|
def create_future(self) -> Future[Any]: ...
|
|
# Tasks methods
|
|
if sys.version_info >= (3, 11):
|
|
@abstractmethod
|
|
def create_task(self, coro: _CoroutineLike[_T], *, name: str | None = None, context: Context | None = None) -> Task[_T]: ... # type: ignore
|
|
else:
|
|
@abstractmethod
|
|
def create_task(self, coro: _CoroutineLike[_T], *, name: str | None = None) -> Task[_T]: ...
|
|
|
|
@abstractmethod
|
|
def set_task_factory(self, factory: _TaskFactory | None) -> None: ...
|
|
@abstractmethod
|
|
def get_task_factory(self) -> _TaskFactory | None: ...
|
|
# Methods for interacting with threads
|
|
if sys.version_info >= (3, 9): # "context" added in 3.9.10/3.10.2
|
|
@abstractmethod
|
|
def call_soon_threadsafe(
|
|
self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts], context: Context | None = None # type: ignore
|
|
) -> Handle: ...
|
|
else:
|
|
@abstractmethod
|
|
def call_soon_threadsafe(self, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> Handle: ...
|
|
|
|
@abstractmethod
|
|
def run_in_executor(self, executor: Any, func: Callable[[Unpack[_Ts]], _T], *args: Unpack[_Ts]) -> Future[_T]: ...
|
|
@abstractmethod
|
|
def set_default_executor(self, executor: Any) -> None: ...
|
|
# Network I/O methods returning Futures.
|
|
@abstractmethod
|
|
async def getaddrinfo(
|
|
self,
|
|
host: bytes | str | None,
|
|
port: bytes | str | int | None,
|
|
*,
|
|
family: int = 0,
|
|
type: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
) -> list[tuple[AddressFamily, SocketKind, int, str, tuple[str, int] | tuple[str, int, int, int]]]: ...
|
|
@abstractmethod
|
|
async def getnameinfo(self, sockaddr: tuple[str, int] | tuple[str, int, int, int], flags: int = 0) -> tuple[str, str]: ...
|
|
if sys.version_info >= (3, 11):
|
|
@overload
|
|
@abstractmethod
|
|
async def create_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
host: str = ...,
|
|
port: int = ...,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
family: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
sock: None = None,
|
|
local_addr: tuple[str, int] | None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
happy_eyeballs_delay: float | None = None,
|
|
interleave: int | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
@overload
|
|
@abstractmethod
|
|
async def create_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
host: None = None,
|
|
port: None = None,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
family: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
sock: socket,
|
|
local_addr: None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
happy_eyeballs_delay: float | None = None,
|
|
interleave: int | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
else:
|
|
@overload
|
|
@abstractmethod
|
|
async def create_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
host: str = ...,
|
|
port: int = ...,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
family: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
sock: None = None,
|
|
local_addr: tuple[str, int] | None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
happy_eyeballs_delay: float | None = None,
|
|
interleave: int | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
@overload
|
|
@abstractmethod
|
|
async def create_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
host: None = None,
|
|
port: None = None,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
family: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
sock: socket,
|
|
local_addr: None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
happy_eyeballs_delay: float | None = None,
|
|
interleave: int | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
|
|
if sys.version_info >= (3, 13):
|
|
# 3.13 added `keep_alive`.
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: str | Sequence[str] | None = None,
|
|
port: int = ...,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: None = None,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
keep_alive: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: None = None,
|
|
port: None = None,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: socket = ...,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
keep_alive: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
elif sys.version_info >= (3, 11):
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: str | Sequence[str] | None = None,
|
|
port: int = ...,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: None = None,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: None = None,
|
|
port: None = None,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: socket = ...,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
else:
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: str | Sequence[str] | None = None,
|
|
port: int = ...,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: None = None,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
@overload
|
|
@abstractmethod
|
|
async def create_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
host: None = None,
|
|
port: None = None,
|
|
*,
|
|
family: int = ...,
|
|
flags: int = ...,
|
|
sock: socket = ...,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
|
|
if sys.version_info >= (3, 11):
|
|
@abstractmethod
|
|
async def start_tls(
|
|
self,
|
|
transport: WriteTransport,
|
|
protocol: BaseProtocol,
|
|
sslcontext: ssl.SSLContext,
|
|
*,
|
|
server_side: bool = False,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
) -> Transport | None: ...
|
|
async def create_unix_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
path: StrPath | None = None,
|
|
*,
|
|
sock: socket | None = None,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
else:
|
|
@abstractmethod
|
|
async def start_tls(
|
|
self,
|
|
transport: BaseTransport,
|
|
protocol: BaseProtocol,
|
|
sslcontext: ssl.SSLContext,
|
|
*,
|
|
server_side: bool = False,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
) -> Transport | None: ...
|
|
async def create_unix_server(
|
|
self,
|
|
protocol_factory: _ProtocolFactory,
|
|
path: StrPath | None = None,
|
|
*,
|
|
sock: socket | None = None,
|
|
backlog: int = 100,
|
|
ssl: _SSLContext = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
start_serving: bool = True,
|
|
) -> Server: ...
|
|
|
|
if sys.version_info >= (3, 11):
|
|
async def connect_accepted_socket(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
sock: socket,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
elif sys.version_info >= (3, 10):
|
|
async def connect_accepted_socket(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
sock: socket,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
if sys.version_info >= (3, 11):
|
|
async def create_unix_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
path: str | None = None,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
sock: socket | None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
ssl_shutdown_timeout: float | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
else:
|
|
async def create_unix_connection(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
path: str | None = None,
|
|
*,
|
|
ssl: _SSLContext = None,
|
|
sock: socket | None = None,
|
|
server_hostname: str | None = None,
|
|
ssl_handshake_timeout: float | None = None,
|
|
) -> tuple[Transport, _ProtocolT]: ...
|
|
|
|
@abstractmethod
|
|
async def sock_sendfile(
|
|
self, sock: socket, file: IO[bytes], offset: int = 0, count: int | None = None, *, fallback: bool | None = None
|
|
) -> int: ...
|
|
@abstractmethod
|
|
async def sendfile(
|
|
self, transport: WriteTransport, file: IO[bytes], offset: int = 0, count: int | None = None, *, fallback: bool = True
|
|
) -> int: ...
|
|
@abstractmethod
|
|
async def create_datagram_endpoint(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
local_addr: tuple[str, int] | str | None = None,
|
|
remote_addr: tuple[str, int] | str | None = None,
|
|
*,
|
|
family: int = 0,
|
|
proto: int = 0,
|
|
flags: int = 0,
|
|
reuse_address: bool | None = None,
|
|
reuse_port: bool | None = None,
|
|
allow_broadcast: bool | None = None,
|
|
sock: socket | None = None,
|
|
) -> tuple[DatagramTransport, _ProtocolT]: ...
|
|
# Pipes and subprocesses.
|
|
@abstractmethod
|
|
async def connect_read_pipe(self, protocol_factory: Callable[[], _ProtocolT], pipe: Any) -> tuple[ReadTransport, _ProtocolT]: ...
|
|
@abstractmethod
|
|
async def connect_write_pipe(self, protocol_factory: Callable[[], _ProtocolT], pipe: Any) -> tuple[WriteTransport, _ProtocolT]: ...
|
|
@abstractmethod
|
|
async def subprocess_shell(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
cmd: bytes | str,
|
|
*,
|
|
stdin: int | IO[Any] | None = -1,
|
|
stdout: int | IO[Any] | None = -1,
|
|
stderr: int | IO[Any] | None = -1,
|
|
universal_newlines: Literal[False] = False,
|
|
shell: Literal[True] = True,
|
|
bufsize: Literal[0] = 0,
|
|
encoding: None = None,
|
|
errors: None = None,
|
|
text: Literal[False] | None = ...,
|
|
**kwargs: Any,
|
|
) -> tuple[SubprocessTransport, _ProtocolT]: ...
|
|
@abstractmethod
|
|
async def subprocess_exec(
|
|
self,
|
|
protocol_factory: Callable[[], _ProtocolT],
|
|
program: Any,
|
|
*args: Any,
|
|
stdin: int | IO[Any] | None = -1,
|
|
stdout: int | IO[Any] | None = -1,
|
|
stderr: int | IO[Any] | None = -1,
|
|
universal_newlines: Literal[False] = False,
|
|
shell: Literal[False] = False,
|
|
bufsize: Literal[0] = 0,
|
|
encoding: None = None,
|
|
errors: None = None,
|
|
**kwargs: Any,
|
|
) -> tuple[SubprocessTransport, _ProtocolT]: ...
|
|
@abstractmethod
|
|
def add_reader(self, fd: FileDescriptorLike, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
|
|
@abstractmethod
|
|
def remove_reader(self, fd: FileDescriptorLike) -> bool: ...
|
|
@abstractmethod
|
|
def add_writer(self, fd: FileDescriptorLike, callback: Callable[[Unpack[_Ts]], Any], *args: Unpack[_Ts]) -> None: ...
|
|
@abstractmethod
|
|
def remove_writer(self, fd: FileDescriptorLike) -> bool: ...
|
|
@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: ReadableBuffer) -> 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) -> tuple[bytes, _RetAddress]: ...
|
|
@abstractmethod
|
|
async def sock_recvfrom_into(self, sock: socket, buf: WriteableBuffer, nbytes: int = 0) -> tuple[int, _RetAddress]: ...
|
|
@abstractmethod
|
|
async def sock_sendto(self, sock: socket, data: ReadableBuffer, address: _Address) -> int: ...
|
|
# Signal handling.
|
|
@abstractmethod
|
|
def add_signal_handler(self, sig: int, callback: Callable[[Unpack[_Ts]], object], *args: Unpack[_Ts]) -> None: ...
|
|
@abstractmethod
|
|
def remove_signal_handler(self, sig: int) -> bool: ...
|
|
# Error handlers.
|
|
@abstractmethod
|
|
def set_exception_handler(self, handler: _ExceptionHandler | None) -> None: ...
|
|
@abstractmethod
|
|
def get_exception_handler(self) -> _ExceptionHandler | None: ...
|
|
@abstractmethod
|
|
def default_exception_handler(self, context: _Context) -> None: ...
|
|
@abstractmethod
|
|
def call_exception_handler(self, context: _Context) -> None: ...
|
|
# Debug flag management.
|
|
@abstractmethod
|
|
def get_debug(self) -> bool: ...
|
|
@abstractmethod
|
|
def set_debug(self, enabled: bool) -> None: ...
|
|
if sys.version_info >= (3, 9):
|
|
@abstractmethod
|
|
async def shutdown_default_executor(self) -> None: ...
|
|
|
|
class AbstractEventLoopPolicy:
|
|
@abstractmethod
|
|
def get_event_loop(self) -> AbstractEventLoop: ...
|
|
@abstractmethod
|
|
def set_event_loop(self, loop: AbstractEventLoop | None) -> None: ...
|
|
@abstractmethod
|
|
def new_event_loop(self) -> AbstractEventLoop: ...
|
|
# Child processes handling (Unix only).
|
|
if sys.version_info < (3, 14):
|
|
if sys.version_info >= (3, 12):
|
|
@abstractmethod
|
|
@deprecated("Deprecated as of Python 3.12; will be removed in Python 3.14")
|
|
def get_child_watcher(self) -> AbstractChildWatcher: ...
|
|
@abstractmethod
|
|
@deprecated("Deprecated as of Python 3.12; will be removed in Python 3.14")
|
|
def set_child_watcher(self, watcher: AbstractChildWatcher) -> None: ...
|
|
else:
|
|
@abstractmethod
|
|
def get_child_watcher(self) -> AbstractChildWatcher: ...
|
|
@abstractmethod
|
|
def set_child_watcher(self, watcher: AbstractChildWatcher) -> None: ...
|
|
|
|
class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy, metaclass=ABCMeta):
|
|
def get_event_loop(self) -> AbstractEventLoop: ...
|
|
def set_event_loop(self, loop: AbstractEventLoop | None) -> None: ...
|
|
def new_event_loop(self) -> AbstractEventLoop: ...
|
|
|
|
def get_event_loop_policy() -> AbstractEventLoopPolicy: ...
|
|
def set_event_loop_policy(policy: AbstractEventLoopPolicy | None) -> None: ...
|
|
def set_event_loop(loop: AbstractEventLoop | None) -> None: ...
|
|
def new_event_loop() -> AbstractEventLoop: ...
|
|
|
|
if sys.version_info < (3, 14):
|
|
if sys.version_info >= (3, 12):
|
|
@deprecated("Deprecated as of Python 3.12; will be removed in Python 3.14")
|
|
def get_child_watcher() -> AbstractChildWatcher: ...
|
|
@deprecated("Deprecated as of Python 3.12; will be removed in Python 3.14")
|
|
def set_child_watcher(watcher: AbstractChildWatcher) -> None: ...
|
|
|
|
else:
|
|
def get_child_watcher() -> AbstractChildWatcher: ...
|
|
def set_child_watcher(watcher: AbstractChildWatcher) -> None: ...
|