tessl/pypi-simple-websocket

Simple WebSocket server and client for Python

—

Pending

Overview

Eval results

Files

ASGI Integration

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

Direct ASGI WebSocket support for modern Python web frameworks implementing the ASGI specification. Provides seamless integration with ASGI-compatible frameworks like FastAPI, Starlette, Django Channels, and Quart.

Capabilities

WebSocketASGI Class

ASGI WebSocket wrapper that provides a simplified interface for handling WebSocket connections in ASGI applications. Handles the ASGI WebSocket protocol internally while exposing a clean API for message handling.

class WebSocketASGI:
    def __init__(self, scope, receive, send, subprotocols=None):
        """
        Initialize ASGI WebSocket wrapper.
        
        Parameters:
        - scope: ASGI scope dictionary containing connection details
        - receive: ASGI receive callable for getting messages
        - send: ASGI send callable for sending messages
        - subprotocols: List of supported subprotocols or None
        """

@classmethod
async def accept(cls, scope, receive, send, subprotocols=None):
    """
    Accept ASGI WebSocket connection.
    
    Parameters:
    - scope: ASGI scope dictionary with connection metadata
    - receive: ASGI receive callable to get messages from client
    - send: ASGI send callable to send messages to client
    - subprotocols: List of supported subprotocols or None for no negotiation
    
    Returns:
    WebSocketASGI instance ready for communication
    """

WebSocketASGI Methods

async def receive(self):
    """
    Receive message from ASGI WebSocket connection.
    
    Returns:
    Received data as str (text messages) or bytes (binary messages)
    
    Raises:
    ConnectionClosed: When client disconnects
    OSError: For unsupported message types
    """

async def send(self, data):
    """
    Send message over ASGI WebSocket connection.
    
    Parameters:
    - data: Data to send (str for text message, bytes for binary message)
    """

async def close(self):
    """
    Close ASGI WebSocket connection.
    
    Sends close message to client through ASGI interface.
    """

WebSocketASGI Attributes

class WebSocketASGI:
    subprotocol: str | None  # Negotiated subprotocol name
    connected: bool          # Connection status

Usage Examples

FastAPI Integration

from fastapi import FastAPI, WebSocket
import simple_websocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    # Accept WebSocket connection via ASGI
    scope = websocket.scope
    receive = websocket.receive
    send = websocket.send
    
    ws = await simple_websocket.WebSocketASGI.accept(
        scope=scope,
        receive=receive,
        send=send,
        subprotocols=['chat', 'echo']
    )
    
    try:
        while True:
            # Receive message from client
            message = await ws.receive()
            print(f"Received: {message}")
            
            # Echo message back
            await ws.send(f"Echo: {message}")
            
    except simple_websocket.ConnectionClosed:
        print("Client disconnected")
    finally:
        await ws.close()

Starlette Integration

from starlette.applications import Starlette
from starlette.websockets import WebSocket
from starlette.routing import WebSocketRoute
import simple_websocket

async def websocket_handler(websocket: WebSocket):
    # Accept WebSocket connection
    ws = await simple_websocket.WebSocketASGI.accept(
        scope=websocket.scope,
        receive=websocket.receive,
        send=websocket.send
    )
    
    try:
        # Send welcome message
        await ws.send("Welcome to the WebSocket server!")
        
        # Handle incoming messages
        while True:
            data = await ws.receive()
            
            # Handle different message types
            if isinstance(data, str):
                await ws.send(f"Text received: {data}")
            elif isinstance(data, bytes):
                await ws.send(f"Binary data received: {len(data)} bytes")
                
    except simple_websocket.ConnectionClosed:
        print("WebSocket connection closed")

routes = [
    WebSocketRoute('/ws', websocket_handler)
]

app = Starlette(routes=routes)

Django Channels Integration

from channels.generic.websocket import AsyncWebsocketConsumer
import simple_websocket

class SimpleWebSocketConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # Accept connection using simple-websocket ASGI wrapper
        self.ws = await simple_websocket.WebSocketASGI.accept(
            scope=self.scope,
            receive=self.receive,
            send=self.send
        )
        
        # Send initial message
        await self.ws.send("Connected to Django Channels via simple-websocket")
    
    async def disconnect(self, close_code):
        if hasattr(self, 'ws'):
            await self.ws.close()
    
    async def receive(self, text_data=None, bytes_data=None):
        try:
            # Use simple-websocket receive method
            data = await self.ws.receive()
            
            # Process and respond
            if isinstance(data, str):
                response = f"Processed text: {data.upper()}"
            else:
                response = f"Processed binary data: {len(data)} bytes"
            
            await self.ws.send(response)
            
        except simple_websocket.ConnectionClosed:
            await self.close()

Custom ASGI Application

import simple_websocket

async def websocket_app(scope, receive, send):
    """
    Simple ASGI WebSocket application using simple-websocket.
    """
    if scope['type'] != 'websocket':
        # Not a WebSocket connection
        await send({
            'type': 'http.response.start',
            'status': 404,
            'headers': []
        })
        await send({
            'type': 'http.response.body',
            'body': b'Not Found'
        })
        return
    
    # Handle WebSocket connection
    ws = await simple_websocket.WebSocketASGI.accept(
        scope=scope,
        receive=receive,
        send=send,
        subprotocols=['v1', 'v2']  # Subprotocol negotiation
    )
    
    try:
        # Send connection info
        await ws.send(f"Connected with subprotocol: {ws.subprotocol}")
        
        # Message handling loop
        message_count = 0
        while True:
            data = await ws.receive()
            message_count += 1
            
            # Send response with message counter
            response = {
                'message_id': message_count,
                'echo': data,
                'subprotocol': ws.subprotocol
            }
            
            await ws.send(str(response))
            
    except simple_websocket.ConnectionClosed:
        print(f"Connection closed after {message_count} messages")

# Run with any ASGI server (uvicorn, hypercorn, etc.)
# uvicorn app:websocket_app --host 0.0.0.0 --port 8000

Subprotocol Negotiation with ASGI

import simple_websocket
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_with_subprotocols(websocket: WebSocket):
    # Custom subprotocol list
    supported_protocols = ['chat.v1', 'chat.v2', 'echo']
    
    ws = await simple_websocket.WebSocketASGI.accept(
        scope=websocket.scope,
        receive=websocket.receive,
        send=websocket.send,
        subprotocols=supported_protocols
    )
    
    try:
        # Handle different protocols differently
        if ws.subprotocol == 'chat.v1':
            await ws.send("Welcome to Chat v1!")
            # Handle v1 protocol specifics
        elif ws.subprotocol == 'chat.v2':
            await ws.send("Welcome to Chat v2 with enhanced features!")
            # Handle v2 protocol specifics
        elif ws.subprotocol == 'echo':
            await ws.send("Echo mode activated")
            # Simple echo functionality
        else:
            await ws.send("No subprotocol selected")
        
        # Main message loop
        while True:
            message = await ws.receive()
            
            # Protocol-specific message handling
            if ws.subprotocol == 'echo':
                await ws.send(f"Echo: {message}")
            else:
                await ws.send(f"[{ws.subprotocol}] Received: {message}")
                
    except simple_websocket.ConnectionClosed:
        print(f"Connection with protocol '{ws.subprotocol}' closed")

Error Handling

import simple_websocket
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def error_handling_example(websocket: WebSocket):
    try:
        ws = await simple_websocket.WebSocketASGI.accept(
            websocket.scope,
            websocket.receive,
            websocket.send
        )
        
        while True:
            try:
                message = await ws.receive()
                await ws.send(f"Received: {message}")
                
            except simple_websocket.ConnectionClosed:
                print("Client disconnected normally")
                break
            except OSError as e:
                print(f"WebSocket error: {e}")
                break
                
    except Exception as e:
        print(f"Failed to establish WebSocket connection: {e}")
        # Connection was not properly established
    
    finally:
        # Cleanup if needed
        try:
            await ws.close()
        except:
            pass  # Connection may already be closed

Install with Tessl CLI