picows

The on_ws_* methods in WSListener are non-async to maximize performance (avoiding event loop overhead and enabling zero-copy). If you need async capabilities, you have two main patterns:

1. Eager Tasks (Recommended for Python 3.13+): Use asyncio.task.eager_task_factory. This allows non-async callbacks to execute async logic immediately without waiting for the next event loop cycle, minimizing latency.
2. Asyncio Queue: If you require an async receive_message() style interface (similar to aiohttp), use an asyncio.Queue. Note that this introduces higher latency because coroutines are only woken up on the next event loop cycle and payloads must be copied.

It is often best to keep the data path non-async and only transition to async when an actual async operation is required.

# Example approach using eager tasks (Python 3.13+)
# This is more efficient than using a Queue for turning callbacks async
import asyncio

# (Implementation details would follow the pattern in echo_client_async_callbacks.py)

The Core API uses a listener-based pattern (WSListener) and a non-async data path for maximum efficiency. Instead of receiving full messages, the user handles WSFrame objects. This allows for zero-copy data handling but requires the user to manage frame concatenation if a message spans multiple frames.

Key components:

• WSListener: A class you subclass to implement event handlers like on_ws_connected and on_ws_frame.
• WSTransport: The object used to interact with the connection (e.g., transport.send(), transport.disconnect()).
• WSFrame: Represents a single WebSocket frame, providing access to the payload and parser state.
• WSMsgType: Enum for message types (e.g., TEXT, CLOSE).
• WSUpgradeRequest: Used in servers to facilitate routing via a listener factory.

# Core Client Example
import asyncio
from picows import ws_connect, WSFrame, WSTransport, WSListener, WSMsgType, WSCloseCode

class ClientListener(WSListener):
    def on_ws_connected(self, transport: WSTransport):
        transport.send(WSMsgType.TEXT, b"Hello world")

    def on_ws_frame(self, transport: WSTransport, frame: WSFrame):
        print(f"Echo reply: {frame.get_payload_as_ascii_text()}")
        transport.send_close(WSCloseCode.OK)
        transport.disconnect()

async def main():
    transport, client = await ws_connect(ClientListener, "ws://127.0.0.1:9001")
    await transport.wait_disconnected()

if __name__ == "__main__":
    asyncio.run(main())

# Core Server Example
import asyncio
from picows import ws_create_server, WSFrame, WSTransport, WSListener, WSMsgType, WSUpgradeRequest

class ServerClientListener(WSListener):
    def on_ws_connected(self, transport: WSTransport):
        print("New client connected")

    def on_ws_frame(self, transport: WSTransport, frame: WSFrame):
        if frame.msg_type == WSMsgType.CLOSE:
            transport.send_close(frame.get_close_code(), frame.get_close_message())
            transport.disconnect()
        else:
            transport.send(frame.msg_type, frame.get_payload_as_memoryview())

async def main():
    def listener_factory(r: WSUpgradeRequest):
        # Routing can be implemented here by analyzing request content
        return ServerClientListener()

    server: asyncio.Server = await ws_create_server(listener_factory, "127.0.0.1", 9001)
    for s in server.sockets:
        print(f"Server started on {s.getsockname()}")

    await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(main())

The websockets API is a drop-in replacement for the websockets library. You only need to change your imports to picows.websockets.asyncio.client or picows.websockets.asyncio.server.

Note: Some features from the original websockets library are not yet supported. Check the official documentation for the full compatibility list.

# Client Example
from picows.websockets.asyncio.client import connect
import asyncio

async def hello():
    async with connect("ws://localhost:8765") as websocket:
        await websocket.send("Hello world!")
        message = await websocket.recv()
        print(message)

if __name__ == "__main__":
    asyncio.run(hello())

# Server Example
from picows.websockets.asyncio.server import serve
import asyncio

async def echo(websocket):
    async for message in websocket:
        await websocket.send(message)

async def main():
    async with serve(echo, "localhost", 8765) as server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(main())

The ws_connect function supports proxies via the proxy argument using a proxy URL.

• HTTP Proxies: Use https://user:password@host:port. When using https://, TLS is established with the proxy first. Use proxy_ssl_context to customize certificate verification.
• SOCKS Proxies: Supports SOCKS4 and SOCKS5. Note that for SOCKS4, hostname resolution is performed by the client, whereas for other protocols, it generally happens at the proxy.
• WSS through Proxy: For wss:// URLs, picows establishes a tunnel through the proxy before performing the TLS handshake with the WebSocket server.
• System Proxies: picows does not use system proxy settings automatically. To use them, retrieve them via urllib.request.getproxies() and pass the result to the proxy argument.

import ssl
from picows import ws_connect

transport, listener = await ws_connect(
    ClientListener,
    "ws://127.0.0.1:9000/",
    proxy_ssl_context=ssl.create_default_context(),
    proxy="https://user:password@127.0.0.1:1080",
)