picows is a high-performance Python library designed for building asyncio WebSocket clients and servers. Implemented in Cython, it offers exceptional speed and efficiency, surpassing other popular Python WebSocket libraries.

The above chart shows the performance of various echo clients communicating with the same high-peformance C++ server through a loopback interface. boost.beast client is also included for reference. You can find benchmark sources and more results here.

• Maximally efficient WebSocket frame parser and builder implemented in Cython
• Reuse memory as much as possible, avoid reallocations, and avoid unnecessary Python object creation
• Use aiofastnet to achieve excellent TCP/TLS performance regardless of the event loop used
• Provide a Cython .pxd for efficient integration of user Cythonized code with picows
• Ability to check if a frame is the last one in the receiving buffer
• Auto ping-pong with an option to customize ping/pong messages
• Convenient method to measure websocket roundtrip time using ping/pong messages

picows requires Python 3.9 or greater and is available on PyPI:

pip install picows

Popular WebSocket libraries provide high-level interfaces that handle timeouts, flow control, optional compression/decompression, and reassembly of WebSocket messages from frames, while also implementing async iteration interfaces. These features are typically implemented in pure Python, resulting in significant overhead even when messages are small, un-fragmented (with every WebSocket frame marked as final), and uncompressed.

The async iteration interface relies on asyncio.Futures, which adds additional work for the event loop, postpone actual message processing and introduce a significant delay. Moreover, it is not always necessary to process every message. In some use cases, only the latest message matters, and previous ones can be discarded without even parsing their content.

The library achieves superior performance by offering an efficient, non-async data path, similar to the transport/protocol design from asyncio. The user handler receives WebSocket frame objects instead of complete messages. Since a message can span multiple frames, it is up to the user to decide the most effective strategy for concatenating them. Each frame object includes additional details about the current parser state, which may help optimize the behavior of the user's application.

Connects to an echo server, sends a message, and disconnects after receiving a reply.

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())

This prints:

Echo reply: Hello world

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_bytes())


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())

1. Fork and clone the repository:

git clone git@github.com:tarasko/picows.git
cd picows

2. Create a virtual environment and activate it:

python3 -m venv picows-dev
source picows-dev/bin/activate

3. Install development dependencies:

# To run tests
pip install -r requirements-test.txt

4. Build in place and run tests:

python setup.py build_ext --inplace --dev
pytest -s -v

# Run specific test with picows debug logs enabled
pytest -s -v -k test_client_handshake_timeout[uvloop-plain] --log-cli-level 9

5. Build coverage report:

pytest -s -v --cov=picows --cov-report=html

6. Build docs:

pip install -r docs/requirements.txt
make -C docs clean html