pyphoenix 

async stop()[source]

Stop the Phoenix application.

Return type:: None

async route_message(socket, message)[source]

Route incoming message to appropriate channel.

Parameters:

socket – The socket that received the message
message (Message) – The parsed message

Return type:

async get_or_create_channel(topic, socket)[source]

Get existing channel or create new one from registered handlers.

Parameters:

topic (str) – The channel topic
socket – The socket requesting the channel

Return type:

Optional[Channel]

Returns:

Channel instance or None if no pattern matches

match_topic_pattern(pattern, topic)[source]

Match topic against pattern (support wildcards).

Parameters:

pattern (str) – Pattern like “room:” or “user:”
topic (str) – Actual topic like “room:lobby” or “user:123”

Return type:

bool

Returns:

True if topic matches pattern

run(host='localhost', port=4000, **kwargs)[source]

Run the Phoenix application (blocking).

Parameters:

host (str) – Host to bind to
port (int) – Port to bind to
**kwargs – Additional arguments (for future use)

Return type:

pyphoenix.create_app()[source]

Create a new Phoenix application instance.

Return type:: Phoenix

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

A Channel represents a conversation on a specific topic between multiple participants.

Channels maintain their own state and handle joining, leaving, pushing messages, and broadcasting to all participants. They support middleware for cross-cutting concerns like authentication, logging, and rate limiting.

The channel lifecycle follows Phoenix conventions:

CLOSED -> JOINING (when join() is called)
JOINING -> JOINED (when join succeeds)
JOINED -> LEAVING (when leave() is called)
LEAVING -> CLOSED (when leave completes)
Any state -> ERRORED (when errors occur)

topic

The channel topic (e.g., “room:lobby”, “user:123”)

Type:: str

params

Parameters passed during channel creation

Type:: dict

socket: Associated socket connection

state

Current channel state

Type:: ChannelState

join_ref

Reference ID for the join operation

Type:: str, optional

timeout

Timeout for operations in seconds (default: 10.0)

Type:: float

middleware

Stack of middleware functions

Type:: MiddlewareStack

Example

Create a basic room channel:

class RoomChannel(Channel):
    async def on_join(self, payload, socket):
        user_id = payload.get("user_id")
        logger.info(f"User {user_id} joined {self.topic}")
        return {"status": "ok"}

    async def on_message(self, payload, socket):
        message = payload.get("text", "")
        await self.broadcast("new_message", {
            "text": message,
            "timestamp": time.time()
        })

Parameters:

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

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

Initialize a channel.

Parameters:

topic (str) – The channel topic (e.g., “room:lobby”)
params (Optional[dict[str, Any]]) – Optional parameters for joining
socket – Associated socket connection

use_middleware(middleware)[source]

Add middleware to the channel’s middleware stack.

Middleware is executed in the order it’s added, with each middleware having the opportunity to process or modify requests before they reach the final handler.

Parameters:: middleware – A middleware instance that implements the required interface (e.g., LoggingMiddleware, AuthMiddleware, RateLimitMiddleware)
Return type:: None

Example

Add authentication and logging middleware:

channel.use_middleware(LoggingMiddleware())
channel.use_middleware(AuthMiddleware(authenticate_user))
channel.use_middleware(RateLimitMiddleware(max_messages=10))

remove_middleware(middleware)[source]

Remove middleware from the channel’s middleware stack.

Return type:: None

async join(params=None)[source]

Join the channel and start receiving messages.

This method initiates the channel join process. If middleware is configured, it will be executed before the final join handler. The method handles authentication, validation, and any other middleware processing.

Parameters:

params (Optional[dict[str, Any]]) – Optional parameters to send with join request. These are merged with the channel’s initial params.

Returns:

An object containing the join status and response data. - status: “ok” if join succeeded, “error” if it failed - response: Dict with additional data (success response or error details)

Return type:

JoinResponse

Raises:

JoinError – If the join operation fails
PyPhoenixTimeoutError – If the join times out

Example

Basic join:

response = await channel.join({"user_id": "123"})
if response.status == "ok":
    print("Successfully joined channel!")

Join with authentication:

response = await channel.join({
    "token": "auth_token",
    "user_id": "123"
})

async leave()[source]

Leave the channel gracefully.

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:

PushResponse

Returns:

PushResponse with status and response data

async broadcast(event, payload)[source]

Broadcast a message to all channel participants.

Parameters:

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

Return type:

async broadcast_from(socket, event, payload)[source]

Broadcast a message to all channel participants except the specified socket.

Parameters:

socket – The socket to exclude from the broadcast
event (str) – The event name
payload (dict[str, Any]) – The message payload

Return type:

async track_presence(user_id, meta)[source]

Track a user’s presence in this channel.

Parameters:

user_id (str) – The user identifier
meta (dict[str, Any]) – Metadata about the presence (username, socket_id, etc.)

Return type:

async untrack_presence(user_id)[source]

Stop tracking a user’s presence in this channel.

Parameters:: user_id (str) – The user identifier
Return type:: None

async list_presence()[source]

List all users currently present in this channel.

Return type:: dict[str, Any]
Returns:: Dictionary mapping user IDs to their presence metadata

async presence_count()[source]

Get the number of users present in this channel.

Return type:: int
Returns:: Number of present users

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 to listen for
callback (Optional[Callable]) – The callback function to invoke (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:

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

Trigger event handlers for an event.

Parameters:

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

Return type:

async handle_event(event, payload, ref, socket)[source]

Handle incoming event from Phoenix routing system.

This is the main entry point for events routed from Phoenix app. Dispatches to appropriate handler methods (on_join, on_message, etc.)

Parameters:

event (str) – The event name (phx_join, message, etc.)
payload (dict[str, Any]) – The event payload
ref (Optional[str]) – Message reference for replies
socket – The socket that sent the message

Return type:

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

Represents a client connection to the Phoenix server.

Handles WebSocket communication, channel management, and message routing.

Parameters:: websocket (Optional[ServerConnection])

__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.

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

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.

Parameters:: raw_message (str) – Raw JSON message from WebSocket
Return type:: None

async close()[source]

Close the socket connection.

Return type:: None

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

WebSocket transport for handling client connections.

Parameters:

host (str)
port (int)

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

Initialize WebSocket transport.

Parameters:

host (str) – Host to bind to
port (int) – Port to bind to

set_phoenix_app(app)[source]

Connect this transport to a Phoenix application.

Parameters:: app – Phoenix application instance

async start()[source]

Start the WebSocket server.

Return type:: None

async stop()[source]

Stop the WebSocket server.

Return type:: None

async handle_connection(websocket)[source]

Handle new WebSocket connection.

Parameters:: websocket (ServerConnection) – The WebSocket connection
Return type:: None

class pyphoenix.PubSub[source]

Local publish-subscribe system for inter-process communication.

This is a simplified, single-node implementation that will later be extended for distributed operation.

__init__()[source]: Initialize the PubSub system.

async subscribe(topic, callback)[source]

Subscribe to a topic.

Parameters:

topic (str) – The topic to subscribe to (supports wildcards like “room:*”)
callback (Callable) – The callback function to invoke for messages

Return type:

Subscription

Returns:

Subscription object that can be used to unsubscribe

async unsubscribe(topic, callback)[source]

Unsubscribe from a topic.

Parameters:

topic (str) – The topic to unsubscribe from
callback (Callable) – The callback function to remove

Return type:

async publish(topic, message)[source]

Publish a message to all subscribers of a topic.

Parameters:

topic (str) – The topic to publish to
message (Any) – The message to send (can be any type)

Return type:

int

Returns:

Number of subscribers that received the message

async direct_publish(node_id, topic, message)[source]

Publish directly to a specific node.

For now, this is the same as regular publish since we only have one node. This method is a placeholder for future distributed functionality.

Parameters:

node_id (str) – Target node identifier (ignored in local mode)
topic (str) – The topic to publish to
message (Any) – The message to send

Return type:

async list_topics()[source]

List all active topics with subscribers.

Return type:: list[str]
Returns:: List of topic names

async subscriber_count(topic)[source]

Get the number of subscribers for a topic.

Parameters:: topic (str) – The topic to check
Return type:: int
Returns:: Number of subscribers

async shutdown()[source]

Shutdown the PubSub system and clear all subscriptions.

Return type:: None

pyphoenix.get_pubsub()[source]

Get or create the global PubSub instance.

Return type:: PubSub

class pyphoenix.Presence[source]

Distributed presence tracking system.

Tracks which processes/users are present on which topics across the system. This is a local implementation that will later be extended for distribution.

__init__()[source]: Initialize the Presence tracker.

async track(pid, topic, meta)[source]

Track a process’s presence on a topic.

Parameters:

pid (str) – Process/user identifier
topic (str) – The topic to track presence on
meta (dict[str, Any]) – Metadata about the presence (e.g., user info, online_at timestamp)

Return type:

async untrack(pid, topic)[source]

Stop tracking a process’s presence.

Parameters:

pid (str) – Process/user identifier
topic (str) – The topic to stop tracking on

Return type:

Optional[PresenceState]

async list(topic)[source]

List all presences for a topic.

Parameters:: topic (str) – The topic to list presences for
Return type:: dict[str, PresenceState]
Returns:: Dictionary mapping process IDs to their presence states

async get_presence(pid, topic)[source]

Get presence state for a specific process on a topic.

Parameters:

pid (str) – Process/user identifier
topic (str) – The topic to check

Return type:

Returns:

PresenceState if present, None otherwise

on_join(callback)[source]

Register callback for when a presence joins.

Parameters:: callback (Callable) – Function called with (topic, pid, presence_state)
Return type:: None

on_leave(callback)[source]

Register callback for when a presence leaves.

Parameters:: callback (Callable) – Function called with (topic, pid, presence_state)
Return type:: None

async sync_presence_state(topic)[source]

Get the current presence state for synchronization.

This method returns the full presence state in a format suitable for sending to new joiners or for synchronization between nodes.

Parameters:: topic (str) – The topic to get state for
Return type:: dict[str, Any]
Returns:: Serializable presence state

async topic_count()[source]

Get the number of topics with active presences.

Return type:: int
Returns:: Number of topics

async total_presence_count()[source]

Get the total number of presences across all topics.

Return type:: int
Returns:: Total presence count

async presence_count_for_topic(topic)[source]

Get the number of presences for a specific topic.

Parameters:: topic (str) – The topic to count presences for
Return type:: int
Returns:: Number of presences on the topic

async cleanup_stale_presences(max_age_seconds=3600)[source]

Clean up stale presences that haven’t been updated recently.

Parameters:: max_age_seconds (float) – Maximum age in seconds before considering a presence stale
Return type:: int
Returns:: Number of stale presences cleaned up

pyphoenix.get_presence()[source]

Get or create the global Presence instance.

Return type:: Presence

class pyphoenix.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

class pyphoenix.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:

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:

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:

class pyphoenix.PhoenixConfig(host='localhost', port=4000, workers=1, debug=False, channel=<factory>, pubsub=<factory>, presence=<factory>, transport=<factory>, security=<factory>, logging=<factory>, custom=<factory>)[source]

Main configuration class for PyPhoenix.

Parameters:

host (str)
port (int)
workers (int)
debug (bool)
channel (ChannelConfig)
pubsub (PubSubConfig)
presence (PresenceConfig)
transport (TransportConfig)
security (SecurityConfig)
logging (LoggingConfig)
custom (dict[str, Any])

host: str = 'localhost'

port: int = 4000

workers: int = 1

debug: bool = False

channel: ChannelConfig

pubsub: PubSubConfig

presence: PresenceConfig

transport: TransportConfig

security: SecurityConfig

logging: LoggingConfig

custom: dict[str, Any]

classmethod from_env()[source]

Create configuration from environment variables.

Return type:: PhoenixConfig

classmethod from_dict(data)[source]

Create configuration from a dictionary.

Parameters:: data (dict[str, Any])
Return type:: PhoenixConfig

to_dict()[source]

Convert configuration to a dictionary.

Return type:: dict[str, Any]

validate()[source]

Validate configuration and return list of errors.

Return type:: list[str]

__init__(host='localhost', port=4000, workers=1, debug=False, channel=<factory>, pubsub=<factory>, presence=<factory>, transport=<factory>, security=<factory>, logging=<factory>, custom=<factory>)

Parameters:

host (str)
port (int)
workers (int)
debug (bool)
channel (ChannelConfig)
pubsub (PubSubConfig)
presence (PresenceConfig)
transport (TransportConfig)
security (SecurityConfig)
logging (LoggingConfig)
custom (dict[str, Any])

pyphoenix.get_config()[source]

Get the global configuration instance.

Return type:: PhoenixConfig

pyphoenix.set_config(config)[source]

Set the global configuration instance.

Parameters:: config (PhoenixConfig)
Return type:: None

class pyphoenix.Middleware[source]

Base class for middleware components.

abstract async process_join(payload, socket, next_handler)[source]

Process channel join requests.

Parameters:

payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

abstract async process_message(event, payload, socket, next_handler)[source]

Process incoming messages.

Parameters:

event (str)
payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

abstract async process_leave(reason, socket, next_handler)[source]

Process channel leave events.

Parameters:

reason (str)
socket (Any)
next_handler (Callable[..., Any])

Return type:

class pyphoenix.LoggingMiddleware(log_level='info')[source]

Middleware that logs all channel activities.

Parameters:: log_level (str)

__init__(log_level='info')[source]

Parameters:: log_level (str)

async process_join(payload, socket, next_handler)[source]

Log join attempts and results.

Parameters:

payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_message(event, payload, socket, next_handler)[source]

Log message processing.

Parameters:

event (str)
payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_leave(reason, socket, next_handler)[source]

Log leave events.

Parameters:

reason (str)
socket (Any)
next_handler (Callable[..., Any])

Return type:

class pyphoenix.AuthMiddleware(auth_handler=None)[source]

Middleware that handles authentication and authorization.

Parameters:: auth_handler (Optional[Callable[[dict[str, Any]], Any]])

__init__(auth_handler=None)[source]

Parameters:: auth_handler (Optional[Callable[[dict[str, Any]], Any]])

async process_join(payload, socket, next_handler)[source]

Authenticate join requests.

Parameters:

payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_message(event, payload, socket, next_handler)[source]

Check authorization for message events.

Parameters:

event (str)
payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_leave(reason, socket, next_handler)[source]

Handle leave cleanup.

Parameters:

reason (str)
socket (Any)
next_handler (Callable[..., Any])

Return type:

class pyphoenix.RateLimitMiddleware(max_messages_per_second=10)[source]

Middleware that implements rate limiting.

Parameters:: max_messages_per_second (int)

__init__(max_messages_per_second=10)[source]

Parameters:: max_messages_per_second (int)

async process_join(payload, socket, next_handler)[source]

No rate limiting on joins.

Parameters:

payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_message(event, payload, socket, next_handler)[source]

Apply rate limiting to messages.

Parameters:

event (str)
payload (dict[str, Any])
socket (Any)
next_handler (Callable[..., Any])

Return type:

async process_leave(reason, socket, next_handler)[source]

Clean up rate limiting data on leave.

Parameters:

reason (str)
socket (Any)
next_handler (Callable[..., Any])

Return type:

class pyphoenix.MiddlewareStack(middleware=None)[source]

Manages and executes middleware in order.

Parameters:: middleware (Optional[list[Middleware]])

__init__(middleware=None)[source]

Parameters:: middleware (Optional[list[Middleware]])

add(middleware)[source]

Add middleware to the stack.

Parameters:: middleware (Middleware)
Return type:: None

remove(middleware)[source]

Remove middleware from the stack.

Parameters:: middleware (Middleware)
Return type:: None

async process_join(payload, socket, final_handler)[source]

Process join through all middleware.

Parameters:

payload (dict[str, Any])
socket (Any)
final_handler (Callable[..., Any])

Return type:

async process_message(event, payload, socket, final_handler)[source]

Process message through all middleware.

Parameters:

event (str)
payload (dict[str, Any])
socket (Any)
final_handler (Callable[..., Any])

Return type:

async process_leave(reason, socket, final_handler)[source]

Process leave through all middleware.

Parameters:

reason (str)
socket (Any)
final_handler (Callable[..., Any])

Return type:

class pyphoenix.MetricsRegistry[source]

Registry for managing metrics.

__init__()[source]

async counter(name, description='')[source]

Get or create a counter metric.

Parameters:

name (str)
description (str)

Return type:

Counter

async gauge(name, description='')[source]

Get or create a gauge metric.

Parameters:

name (str)
description (str)

Return type:

Gauge

async histogram(name, description='', buckets=None)[source]

Get or create a histogram metric.

Parameters:

name (str)
description (str)
buckets (Optional[list[float]])

Return type:

Histogram

async get_all_metrics()[source]

Get all metrics and their values.

Return type:: dict[str, Any]

async clear()[source]

Clear all metrics.

Return type:: None

class pyphoenix.PhoenixMetrics(registry)[source]

Phoenix-specific metrics collector.

Parameters:: registry (MetricsRegistry)

__init__(registry)[source]

Parameters:: registry (MetricsRegistry)

async record_connection(transport='websocket')[source]

Record a new connection.

Parameters:: transport (str)
Return type:: None

async record_disconnection(transport='websocket')[source]

Record a disconnection.

Parameters:: transport (str)
Return type:: None

async record_channel_join(topic, success=True)[source]

Record a channel join attempt.

Parameters:

topic (str)
success (bool)

Return type:

async record_channel_leave(topic)[source]

Record a channel leave.

Parameters:: topic (str)
Return type:: None

async record_message_sent(topic, event)[source]

Record a message sent.

Parameters:

topic (str)
event (str)

Return type:

async record_message_received(topic, event)[source]

Record a message received.

Parameters:

topic (str)
event (str)

Return type:

async record_message_processing_time(topic, event, duration)[source]

Record message processing time.

Parameters:

topic (str)
event (str)
duration (float)

Return type:

async record_presence_track(topic)[source]

Record presence tracking.

Parameters:: topic (str)
Return type:: None

async record_presence_untrack(topic)[source]

Record presence untracking.

Parameters:: topic (str)
Return type:: None

async record_pubsub_publish(topic, subscribers)[source]

Record PubSub message publication.

Parameters:

topic (str)
subscribers (int)

Return type:

async record_error(error_type, component)[source]

Record an error.

Parameters:

error_type (str)
component (str)

Return type:

async pyphoenix.get_metrics_registry()[source]

Get the global metrics registry.

Return type:: MetricsRegistry

async pyphoenix.get_phoenix_metrics()[source]

Get the Phoenix metrics collector.

Return type:: PhoenixMetrics

class pyphoenix.Serializer[source]

Base class for message serializers.

abstract serialize(data)[source]

Serialize data to bytes.

Parameters:: data (Any)
Return type:: bytes

abstract deserialize(data)[source]

Deserialize bytes to data.

Parameters:: data (bytes)
Return type:: Any

abstract property content_type: str: Get the content type for this serializer.

class pyphoenix.JSONSerializer[source]

JSON-based message serializer.

serialize(data)[source]

Serialize data to JSON bytes.

Parameters:: data (Any)
Return type:: bytes

deserialize(data)[source]

Deserialize JSON bytes to data.

Parameters:: data (bytes)
Return type:: Any

property content_type: str: Get the content type for this serializer.

class pyphoenix.MessagePackSerializer[source]

MessagePack-based message serializer.

__init__()[source]

serialize(data)[source]

Serialize data to MessagePack bytes.

Parameters:: data (Any)
Return type:: bytes

deserialize(data)[source]

Deserialize MessagePack bytes to data.

Parameters:: data (bytes)
Return type:: Any

property content_type: str: Get the content type for this serializer.

class pyphoenix.SerializerRegistry[source]

Registry for managing message serializers.

__init__()[source]

register(name, serializer)[source]

Register a serializer.

Parameters:

name (str)
serializer (Serializer)

Return type: