Introduction to wsproto

wsproto is a Python library designed for managing WebSocket connections. It is widely appreciated for its simplicity and high performance, making it an excellent choice for developers seeking to implement WebSocket communications in their applications.

Getting Started with wsproto

Install wsproto using pip:

pip install wsproto

Creating a WebSocket Connection

Here’s an example of how to establish a basic WebSocket connection:

from wsproto import WSConnection
from wsproto.handshake import ServerHandshake

# Create a new connection instance
connection = WSConnection(ServerHandshake)

# Perform the handshake
connection.receive_data(b'...')
response = connection.bytes_to_send()

Sending and Receiving Messages

After the connection has been established, you can send and receive messages:

from wsproto.events import TextMessage, BytesMessage

# Sending a text message
message = TextMessage(data="Hello, World!")
connection.send(message)
outgoing_data = connection.bytes_to_send()

# Receiving a text message
event = connection.receive_data(incoming_data)
if isinstance(event, TextMessage):
    print(f'Received: {event.data}')

Handling Binary Data

Working with binary messages is straightforward with wsproto:

from wsproto.events import BinaryMessage

# Sending a binary message
binary_message = BinaryMessage(data=b'\x00\x01\x02')
connection.send(binary_message)
outgoing_binary_data = connection.bytes_to_send()

# Receiving a binary message
event = connection.receive_data(incoming_binary_data)
if isinstance(event, BinaryMessage):
    print(f'Received binary data: {event.data}')

Closing the Connection

Gracefully closing a WebSocket connection can be done using the following approach:

from wsproto.events import CloseConnection

# Sending a close connection request
close_event = CloseConnection(code=1000, reason='Normal closure')
connection.send(close_event)
outgoing_close_data = connection.bytes_to_send()

# Receiving a close connection request
event = connection.receive_data(incoming_close_data)
if isinstance(event, CloseConnection):
    print(f'Connection closed: {event.reason}')

Example WebSocket App

Here is an example of a simple WebSocket echo server using wsproto:

import socket
from wsproto import WSConnection
from wsproto.handshake import ServerHandshake
from wsproto.events import Request, AcceptConnection, TextMessage, CloseConnection

# Create a TCP/IP socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 8765))
server_socket.listen(1)

while True:
    client_socket, client_address = server_socket.accept()
    connection = WSConnection(ServerHandshake)
    data = client_socket.recv(1024)
    request = connection.receive_data(data)

    if isinstance(request, Request):
        response = AcceptConnection(request)
        connection.send(response)
        client_socket.send(connection.bytes_to_send())

    while True:
        data = client_socket.recv(1024)
        event = connection.receive_data(data)

        if isinstance(event, TextMessage):
            echo_message = TextMessage(data=event.data)
            connection.send(echo_message)
            client_socket.send(connection.bytes_to_send())
        elif isinstance(event, CloseConnection):
            client_socket.close()
            break

This example creates a basic WebSocket server that echoes any received text messages.

With these examples, you are now equipped to utilize wsproto for your WebSocket communication needs in Python. Whether it’s establishing connections or handling various types of messages, wsproto makes it straightforward and efficient.

Hash: ac62f88ca6ed4aeccd7bc3bc756e072c90807676c4da75b8728cd44743ad3429