tessl/pypi-websocket-client

WebSocket client for Python with low level API options

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

WebSocket Client

Name: tessl/pypi-websocket-client
Author: tessl

A comprehensive WebSocket client implementation for Python that enables developers to establish and manage WebSocket connections with low-level API access. It implements the hybi-13 version of the WebSocket protocol and offers both high-level WebSocketApp class for simple use cases and low-level WebSocket class for advanced control over connection handling.

Package Information

Package Name: websocket-client
Language: Python
Installation: pip install websocket-client

Core Imports

import websocket

Common patterns for different use cases:

# High-level event-driven interface
from websocket import WebSocketApp, setReconnect

# Low-level connection control
from websocket import WebSocket, create_connection

# Frame protocol and status codes
from websocket import ABNF, STATUS_NORMAL, STATUS_GOING_AWAY, STATUS_PROTOCOL_ERROR

# Exception handling
from websocket import WebSocketException, WebSocketConnectionClosedException

# Debugging and logging
from websocket import enableTrace

# Socket configuration
from websocket import setdefaulttimeout, getdefaulttimeout

Basic Usage

Simple Connection (Low-level)

from websocket import create_connection

# Connect and exchange messages
ws = create_connection("ws://echo.websocket.events/")
print(ws.recv())  # Server greeting
ws.send("Hello, World")
result = ws.recv()
print(result)  # "Hello, World"
ws.close()

Event-Driven Application (High-level)

import websocket
import rel

def on_message(ws, message):
    print(f"Received: {message}")

def on_error(ws, error):
    print(f"Error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("Connection closed")

def on_open(ws):
    print("Connection opened")

# Create WebSocket application with callbacks
ws = websocket.WebSocketApp("wss://api.gemini.com/v1/marketdata/BTCUSD",
                          on_open=on_open,
                          on_message=on_message,
                          on_error=on_error,
                          on_close=on_close)

# Run with automatic reconnection
ws.run_forever(dispatcher=rel, reconnect=5)
rel.signal(2, rel.abort)
rel.dispatch()

Architecture

The websocket-client library provides a layered architecture for WebSocket communication:

WebSocketApp: High-level event-driven interface with automatic reconnection, ping/pong handling, and lifecycle callbacks
WebSocket: Low-level connection class offering direct protocol control and frame-by-frame communication
ABNF: Frame protocol implementation handling WebSocket message framing, masking, and validation
Connection Functions: Utility functions for establishing connections with full configuration support

This design enables both simple use cases through WebSocketApp and advanced control through the low-level WebSocket class, supporting secure connections (wss://), proxy configurations, custom headers, and comprehensive error handling.

Capabilities

High-Level WebSocket Application

Event-driven WebSocket interface with automatic connection management, reconnection capabilities, and lifecycle callbacks. Ideal for applications requiring persistent connections with minimal configuration.

class WebSocketApp:
    def __init__(
        self,
        url: str,
        header: Union[list, dict, Callable, None] = None,
        on_open: Optional[Callable[[WebSocket], None]] = None,
        on_reconnect: Optional[Callable[[WebSocket], None]] = None,
        on_message: Optional[Callable[[WebSocket, Any], None]] = None,
        on_error: Optional[Callable[[WebSocket, Any], None]] = None,
        on_close: Optional[Callable[[WebSocket, Any, Any], None]] = None,
        on_ping: Optional[Callable] = None,
        on_pong: Optional[Callable] = None,
        on_cont_message: Optional[Callable] = None,
        keep_running: bool = True,
        get_mask_key: Optional[Callable] = None,
        cookie: Optional[str] = None,
        subprotocols: Optional[list] = None,
        on_data: Optional[Callable] = None,
        socket: Optional[socket.socket] = None,
    ) -> None: ...

    def run_forever(
        self,
        sockopt: tuple = None,
        sslopt: dict = None,
        ping_interval: Union[float, int] = 0,
        ping_timeout: Union[float, int, None] = None,
        ping_payload: str = "",
        http_proxy_host: str = None,
        http_proxy_port: Union[int, str] = None,
        http_no_proxy: list = None,
        http_proxy_auth: tuple = None,
        http_proxy_timeout: Optional[float] = None,
        skip_utf8_validation: bool = False,
        host: str = None,
        origin: str = None,
        dispatcher=None,
        suppress_origin: bool = False,
        proxy_type: str = None,
        reconnect: int = None,
    ) -> bool: ...

    def send(self, data: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> None: ...
    def send_text(self, text_data: str) -> None: ...
    def send_bytes(self, data: Union[bytes, bytearray]) -> None: ...
    def close(self, **kwargs) -> None: ...

def setReconnect(reconnectInterval: int) -> None: ...

High-Level WebSocket Application

Low-Level WebSocket Interface

Direct WebSocket protocol control with frame-by-frame communication, custom connection options, and advanced features like manual ping/pong handling and continuation frames.

class WebSocket:
    def __init__(
        self,
        get_mask_key=None,
        sockopt=None,
        sslopt=None,
        fire_cont_frame: bool = False,
        enable_multithread: bool = True,
        skip_utf8_validation: bool = False,
        **_,
    ): ...

    def connect(self, url, **options): ...
    def send(self, payload: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> int: ...
    def send_text(self, text_data: str) -> int: ...
    def send_bytes(self, data: Union[bytes, bytearray]) -> int: ...
    def send_frame(self, frame) -> int: ...
    def recv(self) -> Union[str, bytes]: ...
    def recv_data(self, control_frame: bool = False) -> tuple: ...
    def recv_data_frame(self, control_frame: bool = False) -> tuple: ...
    def ping(self, payload: Union[str, bytes] = ""): ...
    def pong(self, payload: Union[str, bytes] = ""): ...
    def close(self, status: int = STATUS_NORMAL, reason: bytes = b"", timeout: int = 3): ...

def create_connection(url: str, timeout=None, class_=WebSocket, **options) -> WebSocket: ...

Low-Level WebSocket Interface

Frame Protocol and Status Codes

WebSocket frame handling, message types, and protocol status codes for low-level protocol control and custom frame processing.

class ABNF:
    OPCODE_CONT = 0x0
    OPCODE_TEXT = 0x1
    OPCODE_BINARY = 0x2
    OPCODE_CLOSE = 0x8
    OPCODE_PING = 0x9
    OPCODE_PONG = 0xa

    def __init__(
        self,
        fin: int = 0,
        rsv1: int = 0,
        rsv2: int = 0,
        rsv3: int = 0,
        opcode: int = OPCODE_TEXT,
        mask_value: int = 1,
        data: Union[str, bytes, None] = "",
    ) -> None: ...

    def validate(self, skip_utf8_validation: bool = False) -> None: ...
    @staticmethod
    def create_frame(data: Union[bytes, str], opcode: int, fin: int = 1) -> "ABNF": ...
    def format(self) -> bytes: ...
    @staticmethod
    def mask(mask_key: Union[str, bytes], data: Union[str, bytes]) -> bytes: ...

# Status codes
STATUS_NORMAL = 1000
STATUS_GOING_AWAY = 1001
STATUS_PROTOCOL_ERROR = 1002
STATUS_UNSUPPORTED_DATA_TYPE = 1003
STATUS_INVALID_PAYLOAD = 1007
STATUS_POLICY_VIOLATION = 1008
STATUS_MESSAGE_TOO_BIG = 1009
STATUS_INVALID_EXTENSION = 1010
STATUS_UNEXPECTED_CONDITION = 1011

Frame Protocol and Status Codes

Exception Handling

Comprehensive exception hierarchy for WebSocket-specific error conditions, connection management, and protocol violations.

class WebSocketException(Exception): ...
class WebSocketProtocolException(WebSocketException): ...
class WebSocketPayloadException(WebSocketException): ...
class WebSocketConnectionClosedException(WebSocketException): ...
class WebSocketTimeoutException(WebSocketException): ...
class WebSocketProxyException(WebSocketException): ...
class WebSocketBadStatusException(WebSocketException):
    def __init__(
        self,
        message: str,
        status_code: int,
        status_message=None,
        resp_headers=None,
        resp_body=None,
    ): ...
class WebSocketAddressException(WebSocketException): ...

Exception Handling

Debug and Logging

Debugging utilities and logging functions for tracing WebSocket communication, protocol analysis, and troubleshooting connection issues.

def enableTrace(
    traceable: bool,
    handler: logging.StreamHandler = logging.StreamHandler(),
    level: str = "DEBUG",
) -> None: ...

def dump(title: str, message: str) -> None: ...
def error(msg: str) -> None: ...
def warning(msg: str) -> None: ...
def debug(msg: str) -> None: ...
def info(msg: str) -> None: ...
def trace(msg: str) -> None: ...
def isEnabledForError() -> bool: ...
def isEnabledForDebug() -> bool: ...
def isEnabledForTrace() -> bool: ...

Debug and Logging

Socket Configuration

Low-level socket configuration, timeout management, and network-level options for advanced connection control and performance tuning.

DEFAULT_SOCKET_OPTION: list

class sock_opt:
    def __init__(self, sockopt: list, sslopt: dict) -> None: ...

def setdefaulttimeout(timeout: Union[int, float, None]) -> None: ...
def getdefaulttimeout() -> Union[int, float, None]: ...
def recv(sock: socket.socket, bufsize: int) -> bytes: ...
def recv_line(sock: socket.socket) -> bytes: ...
def send(sock: socket.socket, data: Union[bytes, str]) -> int: ...

Socket Configuration

Command-Line Tools

The websocket-client package includes the wsdump command-line utility for interactive WebSocket debugging and testing.

# Install websocket-client to get wsdump command
pip install websocket-client

# Basic usage - connect to WebSocket URL
wsdump ws://echo.websocket.events/

# With proxy support
wsdump --proxy http://127.0.0.1:8080 ws://example.com/ws

# Enable verbose output (show opcodes)
wsdump -v ws://echo.websocket.events/

# Enable full debug tracing
wsdump -vv ws://echo.websocket.events/

# Send initial text message
wsdump --text "Hello Server" ws://echo.websocket.events/

# Ignore SSL certificate errors
wsdump --nocert wss://self-signed.badssl.com/

# Set custom headers
wsdump --headers "Authorization: Bearer token,X-Custom: value" ws://api.example.com/ws

Workspace: tessl
Visibility: Public
Created: 7 months ago
Last updated: 2 months ago
Describes: pkg:pypi/websocket-client@1.8.x
Publish Source: CLI
Badge