tessl/pypi-brython
Python-to-JavaScript transpiler that enables Python 3 development in web browsers with full DOM integration and standard library support
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
websocket.mddocs/
WebSocket Communication
Real-time bidirectional communication with WebSocket support for live data and interactive applications. Provides Python interface to browser WebSocket API for real-time web applications.
Capabilities
WebSocket Connection
Establish and manage WebSocket connections for real-time communication.
class WebSocket:
"""
WebSocket connection handler.
"""
def __init__(self, url: str, protocols: list[str] = None):
"""
Create WebSocket connection.
Args:
url: WebSocket server URL (ws:// or wss://)
protocols: Optional list of sub-protocols
"""
def send(self, data: str | bytes) -> None:
"""
Send data through WebSocket.
Args:
data: Data to send (string or binary)
"""
def close(self, code: int = 1000, reason: str = "") -> None:
"""
Close WebSocket connection.
Args:
code: Close code (1000 = normal closure)
reason: Close reason string
"""
def bind(self, event: str, callback: Callable) -> None:
"""
Bind event handler.
Args:
event: Event type ('open', 'message', 'close', 'error')
callback: Event handler function
"""
# Properties
readyState: int # Connection state (0=connecting, 1=open, 2=closing, 3=closed)
url: str # Connection URL
protocol: str # Selected sub-protocol
bufferedAmount: int # Bytes queued for transmission
# Check WebSocket support
supported: bool # True if WebSocket is available in browserBasic Usage:
from browser.websocket import WebSocket, supported
if not supported:
print("WebSocket not supported in this browser")
else:
# Create WebSocket connection
ws = WebSocket("ws://localhost:8080/chat")
def on_open(event):
print("WebSocket connection established")
ws.send("Hello Server!")
def on_message(event):
message = event.data
print(f"Received: {message}")
def on_close(event):
print(f"Connection closed: {event.code} - {event.reason}")
def on_error(event):
print("WebSocket error occurred")
# Bind event handlers
ws.bind('open', on_open)
ws.bind('message', on_message)
ws.bind('close', on_close)
ws.bind('error', on_error)Chat Application
Real-time chat application using WebSocket communication.
class ChatClient:
"""
Real-time chat client using WebSocket.
"""
def __init__(self, server_url: str, username: str):
"""
Initialize chat client.
Args:
server_url: WebSocket server URL
username: User's display name
"""
def connect(self) -> None:
"""Connect to chat server."""
def disconnect(self) -> None:
"""Disconnect from chat server."""
def send_message(self, message: str) -> None:
"""Send chat message."""
def join_room(self, room: str) -> None:
"""Join chat room."""
def leave_room(self, room: str) -> None:
"""Leave chat room."""Usage:
from browser.websocket import WebSocket
from browser import document, bind
from browser.html import DIV, INPUT, BUTTON, UL, LI
import json
class ChatApp:
def __init__(self):
self.ws = None
self.username = "User"
self.setup_ui()
self.connect()
def setup_ui(self):
"""Create chat interface."""
# Chat container
self.chat_container = DIV(style={
'width': '400px',
'height': '500px',
'border': '1px solid #ccc',
'display': 'flex',
'flex-direction': 'column'
})
# Messages area
self.messages_area = DIV(style={
'flex': '1',
'overflow-y': 'auto',
'padding': '10px',
'background': '#f9f9f9'
})
# Input area
input_area = DIV(style={
'display': 'flex',
'padding': '10px',
'border-top': '1px solid #ccc'
})
self.message_input = INPUT(
placeholder="Type a message...",
style={'flex': '1', 'margin-right': '10px'}
)
send_button = BUTTON("Send")
# Assemble UI
input_area <= self.message_input
input_area <= send_button
self.chat_container <= self.messages_area
self.chat_container <= input_area
document.body <= self.chat_container
# Event handlers
@bind(send_button, 'click')
def handle_send(event):
self.send_message()
@bind(self.message_input, 'keypress')
def handle_keypress(event):
if event.key == 'Enter':
self.send_message()
def connect(self):
"""Connect to WebSocket server."""
self.ws = WebSocket("ws://localhost:8080/chat")
def on_open(event):
self.add_system_message("Connected to chat server")
# Send join message
self.ws.send(json.dumps({
'type': 'join',
'username': self.username
}))
def on_message(event):
try:
data = json.loads(event.data)
self.handle_server_message(data)
except:
self.add_system_message(f"Received: {event.data}")
def on_close(event):
self.add_system_message("Disconnected from server")
def on_error(event):
self.add_system_message("Connection error")
self.ws.bind('open', on_open)
self.ws.bind('message', on_message)
self.ws.bind('close', on_close)
self.ws.bind('error', on_error)
def send_message(self):
"""Send chat message."""
message = self.message_input.value.strip()
if message and self.ws and self.ws.readyState == 1:
self.ws.send(json.dumps({
'type': 'message',
'username': self.username,
'text': message
}))
self.message_input.value = ""
def handle_server_message(self, data):
"""Handle messages from server."""
message_type = data.get('type')
if message_type == 'message':
self.add_chat_message(data['username'], data['text'])
elif message_type == 'join':
self.add_system_message(f"{data['username']} joined the chat")
elif message_type == 'leave':
self.add_system_message(f"{data['username']} left the chat")
elif message_type == 'error':
self.add_system_message(f"Error: {data['message']}")
def add_chat_message(self, username, text):
"""Add chat message to display."""
message_div = DIV(style={'margin': '5px 0'})
username_span = DIV(
f"{username}: ",
style={'font-weight': 'bold', 'display': 'inline'}
)
text_span = DIV(
text,
style={'display': 'inline'}
)
message_div <= username_span
message_div <= text_span
self.messages_area <= message_div
# Scroll to bottom
self.messages_area.scrollTop = self.messages_area.scrollHeight
def add_system_message(self, text):
"""Add system message to display."""
message_div = DIV(
text,
style={
'margin': '5px 0',
'font-style': 'italic',
'color': '#666'
}
)
self.messages_area <= message_div
self.messages_area.scrollTop = self.messages_area.scrollHeight
# Start chat application
chat_app = ChatApp()Live Data Updates
Real-time data streaming for live updates and monitoring.
class LiveDataClient:
"""
Client for receiving live data updates.
"""
def __init__(self, data_url: str):
"""
Initialize live data client.
Args:
data_url: WebSocket URL for data stream
"""
def subscribe(self, channel: str, callback: Callable) -> None:
"""
Subscribe to data channel.
Args:
channel: Data channel name
callback: Function to call with data updates
"""
def unsubscribe(self, channel: str) -> None:
"""Unsubscribe from data channel."""Usage:
from browser.websocket import WebSocket
from browser import document
from browser.html import DIV, TABLE, TR, TD, TH
import json
class StockTicker:
def __init__(self):
self.ws = None
self.stocks = {}
self.setup_ui()
self.connect()
def setup_ui(self):
"""Create stock ticker UI."""
self.container = DIV(style={
'width': '600px',
'margin': '20px auto'
})
title = DIV("Live Stock Prices", style={
'font-size': '24px',
'font-weight': 'bold',
'text-align': 'center',
'margin-bottom': '20px'
})
# Create table
self.table = TABLE(style={
'width': '100%',
'border-collapse': 'collapse'
})
# Table header
header = TR()
header <= TH("Symbol", style={'border': '1px solid #ccc', 'padding': '10px'})
header <= TH("Price", style={'border': '1px solid #ccc', 'padding': '10px'})
header <= TH("Change", style={'border': '1px solid #ccc', 'padding': '10px'})
header <= TH("Volume", style={'border': '1px solid #ccc', 'padding': '10px'})
self.table <= header
self.container <= title
self.container <= self.table
document.body <= self.container
def connect(self):
"""Connect to stock data WebSocket."""
self.ws = WebSocket("ws://localhost:8080/stocks")
def on_open(event):
print("Connected to stock data stream")
# Subscribe to stock updates
self.ws.send(json.dumps({
'action': 'subscribe',
'symbols': ['AAPL', 'GOOGL', 'MSFT', 'TSLA', 'AMZN']
}))
def on_message(event):
try:
data = json.loads(event.data)
if data['type'] == 'quote':
self.update_stock(data)
except Exception as e:
print(f"Error processing data: {e}")
def on_close(event):
print("Stock data stream closed")
self.ws.bind('open', on_open)
self.ws.bind('message', on_message)
self.ws.bind('close', on_close)
def update_stock(self, quote):
"""Update stock display with new quote."""
symbol = quote['symbol']
price = quote['price']
change = quote['change']
volume = quote['volume']
# Store quote
self.stocks[symbol] = quote
# Find or create table row
row_id = f"row_{symbol}"
row = document.get(row_id)
if not row:
# Create new row
row = TR(id=row_id)
symbol_cell = TD(symbol, style={'border': '1px solid #ccc', 'padding': '10px'})
price_cell = TD(f"${price:.2f}", style={'border': '1px solid #ccc', 'padding': '10px'})
# Color change cell based on positive/negative
change_color = 'green' if change >= 0 else 'red'
change_text = f"+{change:.2f}" if change >= 0 else f"{change:.2f}"
change_cell = TD(change_text, style={
'border': '1px solid #ccc',
'padding': '10px',
'color': change_color
})
volume_cell = TD(f"{volume:,}", style={'border': '1px solid #ccc', 'padding': '10px'})
row <= symbol_cell
row <= price_cell
row <= change_cell
row <= volume_cell
self.table <= row
else:
# Update existing row
cells = row.querySelectorAll('td')
cells[1].textContent = f"${price:.2f}"
change_color = 'green' if change >= 0 else 'red'
change_text = f"+{change:.2f}" if change >= 0 else f"{change:.2f}"
cells[2].textContent = change_text
cells[2].style.color = change_color
cells[3].textContent = f"{volume:,}"
# Start stock ticker
stock_ticker = StockTicker()Connection Management
Advanced WebSocket connection handling with reconnection and heartbeat.
class ReliableWebSocket:
"""
WebSocket with automatic reconnection and heartbeat.
"""
def __init__(self, url: str, options: dict = None):
"""
Initialize reliable WebSocket.
Args:
url: WebSocket server URL
options: Connection options (reconnect_interval, max_retries, etc.)
"""
def set_heartbeat(self, interval: int, message: str = "ping") -> None:
"""
Enable heartbeat to keep connection alive.
Args:
interval: Heartbeat interval in milliseconds
message: Heartbeat message to send
"""
def enable_auto_reconnect(self, max_retries: int = -1,
backoff: bool = True) -> None:
"""
Enable automatic reconnection on disconnect.
Args:
max_retries: Maximum reconnection attempts (-1 for unlimited)
backoff: Use exponential backoff for retry delays
"""Usage:
from browser.websocket import WebSocket
from browser.timer import set_interval, clear_interval
class ReliableConnection:
def __init__(self, url):
self.url = url
self.ws = None
self.reconnect_attempts = 0
self.max_reconnect_attempts = 5
self.reconnect_interval = 1000 # Start with 1 second
self.heartbeat_interval = None
self.connect()
def connect(self):
"""Establish WebSocket connection."""
print(f"Connecting to {self.url}...")
self.ws = WebSocket(self.url)
def on_open(event):
print("Connected successfully")
self.reconnect_attempts = 0
self.reconnect_interval = 1000 # Reset interval
self.start_heartbeat()
def on_message(event):
if event.data == "pong":
print("Heartbeat response received")
else:
# Handle actual messages
print(f"Received: {event.data}")
def on_close(event):
print("Connection closed")
self.stop_heartbeat()
if event.code != 1000: # Not normal closure
self.attempt_reconnect()
def on_error(event):
print("Connection error")
self.ws.bind('open', on_open)
self.ws.bind('message', on_message)
self.ws.bind('close', on_close)
self.ws.bind('error', on_error)
def start_heartbeat(self):
"""Start sending heartbeat messages."""
def send_ping():
if self.ws and self.ws.readyState == 1:
self.ws.send("ping")
self.heartbeat_interval = set_interval(send_ping, 30000) # 30 seconds
def stop_heartbeat(self):
"""Stop heartbeat."""
if self.heartbeat_interval:
clear_interval(self.heartbeat_interval)
self.heartbeat_interval = None
def attempt_reconnect(self):
"""Attempt to reconnect with exponential backoff."""
if self.reconnect_attempts >= self.max_reconnect_attempts:
print("Max reconnection attempts reached")
return
self.reconnect_attempts += 1
delay = self.reconnect_interval * (2 ** (self.reconnect_attempts - 1))
print(f"Reconnecting in {delay}ms (attempt {self.reconnect_attempts})")
def do_reconnect():
self.connect()
set_timeout(do_reconnect, delay)
def send(self, data):
"""Send data if connected."""
if self.ws and self.ws.readyState == 1:
self.ws.send(data)
else:
print("Cannot send - not connected")
def close(self):
"""Close connection."""
self.stop_heartbeat()
if self.ws:
self.ws.close(1000, "Client closing")
# Usage
reliable_ws = ReliableConnection("ws://localhost:8080/api")This WebSocket system provides comprehensive real-time communication capabilities for Brython applications, enabling live chat, data streaming, notifications, and other interactive features that require bidirectional communication between client and server.
Install with Tessl CLI
npx tessl i tessl/pypi-brython