or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

asgi-integration.mdasync-websocket.mdexceptions.mdindex.mdsync-websocket.md
tile.json

tessl/pypi-simple-websocket

Simple WebSocket server and client for Python

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/simple-websocket@1.1.x

To install, run

npx @tessl/cli install tessl/pypi-simple-websocket@1.1.0

index.mddocs/

Simple WebSocket

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