pyphoenix.socket

Socket and WebSocket transport implementation for PyPhoenix.

Classes

`Socket`([websocket, phoenix_app])	Represents a client connection to the Phoenix server.
`WebSocketTransport`([host, port])	WebSocket transport for handling client connections.

class pyphoenix.socket.Socket(websocket=None, phoenix_app=None)[source]

Represents a client connection to the Phoenix server.

Handles WebSocket communication, channel management, and message routing.

__init__(websocket=None, phoenix_app=None)[source]

Initialize a socket connection.

Parameters:

websocket (Optional[ServerConnection]) – The WebSocket connection (None for testing)
phoenix_app – Phoenix application instance for routing

async push(message)[source]

Send a message through the socket.

async channel(topic, params=None)[source]

Get or create a channel for the given topic.

Parameters:

topic (str) – The channel topic
params (Optional[dict[str, Any]]) – Optional parameters for the channel

Return type:

Channel

Returns:

Channel instance

async handle_websocket_message(raw_message)[source]

Handle incoming WebSocket message.

async close()[source]

Close the socket connection.

class pyphoenix.socket.WebSocketTransport(host='localhost', port=4000)[source]

WebSocket transport for handling client connections.

Parameters:

__init__(host='localhost', port=4000)[source]

Initialize WebSocket transport.

Parameters:

set_phoenix_app(app)[source]

Connect this transport to a Phoenix application.

async start()[source]

Start the WebSocket server.

async stop()[source]

Stop the WebSocket server.

async handle_connection(websocket)[source]

Handle new WebSocket connection.