tessl/pypi-simple-websocket

Simple WebSocket server and client for Python

—

Pending

Overview

Eval results

Files

Exception Handling

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

Simple WebSocket provides specific exception classes for handling WebSocket-related errors. These exceptions provide detailed information about connection failures and closures, enabling proper error handling and recovery strategies.

Capabilities

Base Exception Class

Base exception class for all simple-websocket related errors, serving as the parent class for WebSocket-specific exceptions.

class SimpleWebsocketError(RuntimeError):
    """
    Base exception class for all simple-websocket related errors.
    
    Inherits from RuntimeError and serves as the parent class for all
    WebSocket-specific exceptions in the simple-websocket library.
    """

Connection Error Exception

Exception raised when WebSocket connection cannot be established or encounters an error during operation.

class ConnectionError(SimpleWebsocketError):
    """
    Connection error exception class.
    
    Raised when:
    - WebSocket handshake fails
    - Invalid server response during connection
    - Network connectivity issues
    - SSL/TLS certificate errors
    - Server rejection of connection
    """
    
    def __init__(self, status_code=None):
        """
        Initialize connection error.
        
        Parameters:
        - status_code: HTTP status code associated with the error (optional)
        """

Connection Closed Exception

Exception raised when WebSocket connection is closed, either by the remote peer or due to protocol violations.

class ConnectionClosed(SimpleWebsocketError):
    """
    Connection closed exception class.
    
    Raised when:
    - Remote peer closes the connection
    - Connection is closed due to protocol violation
    - Network connection is lost
    - Maximum message size exceeded
    - Ping/pong timeout occurs
    """
    
    def __init__(self, reason=None, message=None):
        """
        Initialize connection closed exception.
        
        Parameters:
        - reason: CloseReason enum value indicating why connection was closed
        - message: Optional text message describing the closure
        """

Exception Attributes

class ConnectionError:
    status_code: int | None  # HTTP status code if available

class ConnectionClosed:
    reason: CloseReason      # Close reason code from WebSocket protocol
    message: str | None      # Optional close message text

Close Reason Codes

WebSocket connections can be closed for various reasons. The ConnectionClosed exception includes a reason code that indicates why the connection was terminated:

Standard Close Codes

1000 (Normal Closure): Connection completed successfully
1001 (Going Away): Server going down or browser navigating away
1002 (Protocol Error): WebSocket protocol error
1003 (Unsupported Data): Unsupported data type received
1005 (No Status Received): No close status code was provided
1006 (Abnormal Closure): Connection closed abnormally
1007 (Invalid Data): Invalid UTF-8 data received
1008 (Policy Violation): Message violates endpoint policy
1009 (Message Too Big): Message too large to process
1010 (Missing Extension): Required extension not negotiated
1011 (Internal Error): Server encountered unexpected condition
1015 (TLS Handshake): TLS handshake failure

Usage Examples

Basic Error Handling

import simple_websocket

def handle_websocket_connection():
    try:
        # Attempt to connect
        ws = simple_websocket.Client.connect("ws://localhost:8000/ws")
        
        # Send and receive messages
        ws.send("Hello, Server!")
        response = ws.receive(timeout=10)
        print(f"Response: {response}")
        
    except simple_websocket.ConnectionError as e:
        print(f"Failed to connect: {e}")
        if e.status_code:
            print(f"HTTP Status: {e.status_code}")
            
    except simple_websocket.ConnectionClosed as e:
        print(f"Connection was closed: {e}")
        print(f"Close reason: {e.reason}")
        if e.message:
            print(f"Close message: {e.message}")
            
    except Exception as e:
        print(f"Unexpected error: {e}")
        
    finally:
        # Always clean up
        try:
            ws.close()
        except:
            pass  # Connection may already be closed

Async Error Handling

import asyncio
import simple_websocket

async def async_websocket_handler():
    ws = None
    try:
        # Attempt async connection
        ws = await simple_websocket.AioClient.connect("wss://secure-server.com/ws")
        
        # Send message and wait for response
        await ws.send("Hello, Async Server!")
        response = await ws.receive(timeout=15)
        print(f"Server response: {response}")
        
    except simple_websocket.ConnectionError as e:
        print(f"Connection failed: {e}")
        if e.status_code == 403:
            print("Access forbidden - check authentication")
        elif e.status_code == 404:
            print("WebSocket endpoint not found")
        elif e.status_code >= 500:
            print("Server error - try again later")
            
    except simple_websocket.ConnectionClosed as e:
        print(f"Connection closed: {e}")
        
        # Handle different close reasons
        if e.reason == 1000:  # Normal closure
            print("Connection closed normally")
        elif e.reason == 1001:  # Going away
            print("Server is shutting down")
        elif e.reason == 1008:  # Policy violation
            print("Message violated server policy")
        elif e.reason == 1009:  # Message too big
            print("Message was too large")
        else:
            print(f"Connection closed with code: {e.reason}")
            
    except asyncio.TimeoutError:
        print("Timeout waiting for server response")
        
    finally:
        if ws:
            await ws.close()

# Run the async handler
asyncio.run(async_websocket_handler())

Server Error Handling

import simple_websocket

def websocket_server_handler(environ, start_response):
    ws = None
    try:
        # Accept WebSocket connection
        ws = simple_websocket.Server.accept(
            environ,
            subprotocols=['chat', 'echo'],
            max_message_size=1024 * 1024  # 1MB limit
        )
        
        print(f"Client connected with subprotocol: {ws.subprotocol}")
        
        while True:
            try:
                # Receive message with timeout
                message = ws.receive(timeout=30)
                
                if message is None:
                    # Timeout occurred
                    print("Client timeout - closing connection")
                    ws.close(reason=1000, message="Timeout")
                    break
                
                # Echo the message
                ws.send(f"Echo: {message}")
                
            except simple_websocket.ConnectionClosed as e:
                print(f"Client disconnected: {e}")
                if e.reason == 1001:
                    print("Client navigated away")
                elif e.reason == 1006:
                    print("Connection lost unexpectedly")
                break
                
    except simple_websocket.ConnectionError as e:
        print(f"Failed to establish WebSocket connection: {e}")
        if e.status_code == 400:
            print("Bad WebSocket request")
        elif e.status_code == 426:
            print("Upgrade required")
            
    except Exception as e:
        print(f"Unexpected server error: {e}")
        
    finally:
        if ws:
            try:
                ws.close(reason=1011, message="Server error")
            except:
                pass  # Connection may already be closed

Custom Exception Handling

import simple_websocket
from wsproto.frame_protocol import CloseReason

class WebSocketManager:
    def __init__(self):
        self.ws = None
        self.reconnect_attempts = 0
        self.max_reconnect_attempts = 3
    
    async def connect_with_retry(self, url):
        """Connect with automatic retry logic."""
        while self.reconnect_attempts < self.max_reconnect_attempts:
            try:
                self.ws = await simple_websocket.AioClient.connect(url)
                self.reconnect_attempts = 0  # Reset on successful connection
                return True
                
            except simple_websocket.ConnectionError as e:
                self.reconnect_attempts += 1
                print(f"Connection attempt {self.reconnect_attempts} failed: {e}")
                
                if e.status_code == 401:
                    print("Authentication required - cannot retry")
                    return False
                elif e.status_code == 404:
                    print("Endpoint not found - cannot retry")
                    return False
                elif self.reconnect_attempts < self.max_reconnect_attempts:
                    await asyncio.sleep(2 ** self.reconnect_attempts)  # Exponential backoff
                    
        print("Max reconnection attempts reached")
        return False
    
    async def handle_message_loop(self):
        """Handle messages with comprehensive error handling."""
        try:
            while True:
                message = await self.ws.receive(timeout=60)
                
                if message is None:
                    # Send ping to keep connection alive
                    await self.ws.send("ping")
                    continue
                
                # Process message
                await self.process_message(message)
                
        except simple_websocket.ConnectionClosed as e:
            print(f"Connection closed: {e}")
            
            # Decide whether to reconnect based on close reason
            should_reconnect = e.reason in [
                CloseReason.GOING_AWAY,          # 1001
                CloseReason.ABNORMAL_CLOSURE,    # 1006
                CloseReason.INTERNAL_ERROR       # 1011
            ]
            
            if should_reconnect:
                print("Attempting to reconnect...")
                return await self.connect_with_retry(self.url)
            else:
                print("Connection closed permanently")
                return False
    
    async def process_message(self, message):
        """Process received message."""
        try:
            # Your message processing logic here
            print(f"Processing: {message}")
            await self.ws.send(f"Processed: {message}")
            
        except Exception as e:
            print(f"Error processing message: {e}")
            # Send error response
            await self.ws.send(f"Error: {str(e)}")

Best Practices

1. Always Handle Both Exception Types

try:
    # WebSocket operations
    pass
except simple_websocket.ConnectionError as e:
    # Handle connection establishment errors
    pass
except simple_websocket.ConnectionClosed as e:
    # Handle connection closure
    pass

2. Check Close Reasons for Appropriate Response

except simple_websocket.ConnectionClosed as e:
    if e.reason in [1000, 1001]:  # Normal closures
        # Clean shutdown
        pass
    elif e.reason in [1002, 1003, 1007]:  # Protocol errors
        # Log error, don't retry
        pass
    elif e.reason in [1006, 1011]:  # Abnormal/server errors
        # Consider reconnection
        pass

3. Use Finally Blocks for Cleanup

try:
    # WebSocket operations
    pass
except simple_websocket.ConnectionError:
    # Handle errors
    pass
finally:
    # Always cleanup resources
    if ws:
        try:
            ws.close()
        except:
            pass  # Ignore cleanup errors

Install with Tessl CLI