tessl/pypi-simple-websocket

Simple WebSocket server and client for Python

—

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

Simple WebSocket

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

Simple WebSocket provides easy-to-use WebSocket server and client implementations for Python applications. It supports both synchronous (blocking) and asynchronous (async/await) programming patterns, with built-in connection management, message handling, and support for multiple web frameworks.

Package Information

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

Core Imports

import simple_websocket

Direct class imports:

from simple_websocket import Server, Client, AioServer, AioClient
from simple_websocket import ConnectionError, ConnectionClosed, SimpleWebsocketError

ASGI integration import:

from simple_websocket.asgi import WebSocketASGI

Basic Usage

Synchronous Server

import simple_websocket

# Accept WebSocket connection from WSGI environ
ws = simple_websocket.Server.accept(environ)

# Send and receive messages
ws.send("Hello from server!")
message = ws.receive()
print(f"Received: {message}")

# Close connection
ws.close()

Synchronous Client

import simple_websocket

# Connect to WebSocket server
ws = simple_websocket.Client.connect("ws://localhost:8000/ws")

# Send and receive messages
ws.send("Hello from client!")
response = ws.receive()
print(f"Server responded: {response}")

# Close connection
ws.close()

Asynchronous Server

import simple_websocket

# Accept async WebSocket connection (aiohttp example)
ws = await simple_websocket.AioServer.accept(aiohttp=request)

# Send and receive messages
await ws.send("Hello from async server!")
message = await ws.receive()
print(f"Received: {message}")

# Close connection
await ws.close()

Asynchronous Client

import simple_websocket

# Connect to WebSocket server asynchronously
ws = await simple_websocket.AioClient.connect("ws://localhost:8000/ws")

# Send and receive messages
await ws.send("Hello from async client!")
response = await ws.receive()
print(f"Server responded: {response}")

# Close connection
await ws.close()

Architecture

Simple WebSocket provides two implementation layers:

Synchronous Layer: Server and Client classes using threads for I/O operations
Asynchronous Layer: AioServer and AioClient classes using asyncio for I/O operations

Both layers share common functionality:

Message Handling: Automatic detection of text vs binary messages based on data type
Connection Management: Built-in connection lifecycle management with proper cleanup
Framework Integration: Support for multiple web frameworks (WSGI, ASGI, aiohttp)
Subprotocol Support: Full WebSocket subprotocol negotiation
Error Handling: Specific exceptions for connection errors and closures

Capabilities

Synchronous WebSocket Operations

Core synchronous WebSocket server and client functionality with thread-based I/O operations. Supports WSGI environments and direct socket connections with automatic message type detection.

class Server:
    @classmethod
    def accept(cls, environ, subprotocols=None, receive_bytes=4096, 
               ping_interval=None, max_message_size=None, thread_class=None, 
               event_class=None, selector_class=None): ...
    
    def send(self, data): ...
    def receive(self, timeout=None): ...
    def close(self, reason=None, message=None): ...

class Client:
    @classmethod
    def connect(cls, url, subprotocols=None, headers=None, receive_bytes=4096,
                ping_interval=None, max_message_size=None, ssl_context=None,
                thread_class=None, event_class=None): ...
    
    def send(self, data): ...
    def receive(self, timeout=None): ...
    def close(self, reason=None, message=None): ...

Synchronous WebSocket

Asynchronous WebSocket Operations

Async/await WebSocket server and client functionality using asyncio. Supports multiple async frameworks including aiohttp, ASGI, and custom socket integration with full async context management.

class AioServer:
    @classmethod
    async def accept(cls, aiohttp=None, asgi=None, sock=None, headers=None,
                     subprotocols=None, receive_bytes=4096, ping_interval=None,
                     max_message_size=None): ...
    
    async def send(self, data): ...
    async def receive(self, timeout=None): ...
    async def close(self, reason=None, message=None): ...

class AioClient:
    @classmethod
    async def connect(cls, url, subprotocols=None, headers=None, receive_bytes=4096,
                      ping_interval=None, max_message_size=None, ssl_context=None): ...
    
    async def send(self, data): ...
    async def receive(self, timeout=None): ...
    async def close(self, reason=None, message=None): ...

Asynchronous WebSocket

ASGI Integration

Direct ASGI WebSocket support for modern Python web frameworks that implement ASGI specification, providing seamless integration with ASGI-compatible frameworks.

class WebSocketASGI:
    @classmethod
    async def accept(cls, scope, receive, send, subprotocols=None): ...
    
    async def receive(self): ...
    async def send(self, data): ...
    async def close(self): ...

ASGI Integration

Exception Handling

WebSocket-specific exceptions for handling connection errors and connection closures with detailed error information and status codes.

class SimpleWebsocketError(RuntimeError): ...

class ConnectionError(SimpleWebsocketError):
    def __init__(self, status_code=None): ...

class ConnectionClosed(SimpleWebsocketError):
    def __init__(self, reason=None, message=None): ...

Exception Handling