from socket import socket
from typing import Any
from _typeshed import ReadableBuffer
STATUS_NORMAL = 1000
class ABNF:
"""
ABNF frame class.
See http://tools.ietf.org/html/rfc5234
and http://tools.ietf.org/html/rfc6455#section-5.2
"""
# operation code values.
OPCODE_CONT = 0x0
OPCODE_TEXT = 0x1
OPCODE_BINARY = 0x2
OPCODE_CLOSE = 0x8
OPCODE_PING = 0x9
OPCODE_PONG = 0xA
# available operation code value tuple
OPCODES = (
OPCODE_CONT,
OPCODE_TEXT,
OPCODE_BINARY,
OPCODE_CLOSE,
OPCODE_PING,
OPCODE_PONG,
)
# opcode human readable string
OPCODE_MAP = {
OPCODE_CONT: "cont",
OPCODE_TEXT: "text",
OPCODE_BINARY: "binary",
OPCODE_CLOSE: "close",
OPCODE_PING: "ping",
OPCODE_PONG: "pong",
}
# data length threshold.
LENGTH_7 = 0x7E
LENGTH_16 = 1 << 16
LENGTH_63 = 1 << 63
class WebSocketException(Exception): ...
class WebSocketConnectionClosedException(WebSocketException): ...
class WebSocket:
sock: socket
def send(self, payload: str, opcode: int = ABNF.OPCODE_TEXT) -> int: ...
def ping(self, payload: str = ...) -> None: ...
def recv_data(
self, control_frame: bool
) -> tuple[int, Any,]: ...
def close(
self, status: int = STATUS_NORMAL, reason: bytes = b"", timeout: int = 3
) -> None: ...
def create_connection(
url: str,
timeout: int | float | None = ...,
class_: type[Any] = WebSocket,
header: list[str] | dict[str, str] | None = ...,
cookie: str | None = ...,
origin: str | None = ...,
suppress_origin: bool | None = ...,
host: str | None = ...,
proxy_type: str | None = ...,
http_proxy_host: str | None = ...,
http_proxy_port: str | int | None = ...,
http_no_proxy: list[str] | None = ...,
http_proxy_auth: tuple[str, str] | None = ...,
http_proxy_timeout: int | float | None = ...,
enable_multithread: bool | None = ...,
redirect_limit: int | None = ...,
sockopt: tuple[int, int, int | ReadableBuffer]
| tuple[int, int, None, int]
| None = ...,
sslopt: dict[str, Any] | None = ...,
subprotocols: list[str] | None = ...,
skip_utf8_validation: bool | None = ...,
socket: socket | None = ...,
) -> WebSocket: ...
