pyphoenix.client

Client-side Socket implementation for PyPhoenix.

Classes

`ClientChannel`(topic, socket[, params])	Client-side channel implementation.
`ClientSocket`(url[, params])	Client-side Phoenix socket implementation.
`ConnectionState`(value[, names, module, ...])	Client connection states.

class pyphoenix.client.ConnectionState(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Client connection states.

DISCONNECTED = 'disconnected'

CONNECTING = 'connecting'

CONNECTED = 'connected'

RECONNECTING = 'reconnecting'

CLOSING = 'closing'

CLOSED = 'closed'

class pyphoenix.client.ClientChannel(topic, socket, params=None)[source]

Client-side channel implementation.

Parameters:

topic (str)
socket (ClientSocket)
params (Optional[dict[str, Any]])

__init__(topic, socket, params=None)[source]

Initialize a client channel.

Parameters:

topic (str) – The channel topic
socket (ClientSocket) – The client socket
params (Optional[dict[str, Any]]) – Optional join parameters

async join(timeout=None)[source]

Join the channel.

Parameters:: timeout (Optional[float]) – Optional join timeout
Return type:: dict[str, Any]
Returns:: Join response from server

async leave()[source]

Leave the channel.

Return type:: None

async push(event, payload)[source]

Push a message to the channel.

Parameters:

event (str) – The event name
payload (dict[str, Any]) – The message payload

Return type:

None

on(event, callback=None)[source]

Register an event handler.

Can be used as a decorator:: @channel.on(“event_name”) def handler(payload, ref):

pass
Or as a regular method:: channel.on(“event_name”, handler)

Parameters:

event (str) – The event name
callback (Optional[Callable]) – The callback function (optional for decorator usage)

off(event, callback=None)[source]

Remove event handler(s).

Parameters:

event (str) – The event name
callback (Optional[Callable]) – Specific callback to remove, or None to remove all

Return type:

None

async trigger(event, payload, ref=None)[source]

Trigger event handlers.

Parameters:

event (str) – The event name
payload (dict[str, Any]) – The event payload
ref (Optional[str]) – Optional message reference

Return type:

None

class pyphoenix.client.ClientSocket(url, params=None)[source]

Client-side Phoenix socket implementation.

Parameters:

url (str)
params (Optional[dict[str, Any]])

__init__(url, params=None)[source]

Initialize a client socket.

Parameters:

url (str) – WebSocket URL to connect to
params (Optional[dict[str, Any]]) – Optional connection parameters

async connect(timeout=10.0)[source]

Connect to the Phoenix server.

Parameters:: timeout (float) – Connection timeout in seconds
Return type:: None

async disconnect()[source]

Disconnect from the Phoenix server.

Return type:: None

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

Get or create a channel.

Parameters:

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

Return type:

ClientChannel

Returns:

ClientChannel instance

async push(message)[source]

Push a message to the server.

Parameters:: message (Message) – The message to send
Return type:: None

on_connect(handler)[source]

Register a connect event handler.

Parameters:: handler (Callable)
Return type:: None

on_disconnect(handler)[source]

Register a disconnect event handler.

Parameters:: handler (Callable)
Return type:: None

on_error(handler)[source]

Register an error event handler.

Parameters:: handler (Callable)
Return type:: None