pyphoenix.channel
Channel implementation for PyPhoenix.
This module provides the core Channel class which implements Phoenix-style real-time communication channels. Channels support joining, leaving, pushing messages, broadcasting to all participants, and middleware processing.
- Key features:
Phoenix wire format compatibility
Middleware support for cross-cutting concerns
Automatic rejoining and error handling
Event-based message routing
State management with proper lifecycle hooks
Example
Basic channel usage:
from pyphoenix import Channel
class RoomChannel(Channel):
async def on_join(self, payload, socket):
return {"status": "ok", "response": {"message": "Welcome!"}}
async def on_message(self, payload, socket):
await self.broadcast("new_message", payload)
With middleware:
class AuthenticatedRoomChannel(Channel):
def __init__(self, topic, params=None, socket=None):
super().__init__(topic, params, socket)
self.use_middleware(LoggingMiddleware())
self.use_middleware(AuthMiddleware(authenticate_user))
Classes
|
A Channel represents a conversation on a specific topic between multiple participants. |
- class pyphoenix.channel.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)
- socket
Associated socket connection
- state
Current channel state
- Type:
- middleware
Stack of middleware functions
- Type:
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() })
- 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:
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:
- 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:
- 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 push(event, payload)[source]
Push a message to the channel.
- Parameters:
- Return type:
- Returns:
PushResponse with status and response data
- async broadcast_from(socket, event, payload)[source]
Broadcast a message to all channel participants except the specified socket.
- async presence_count()[source]
Get the number of users present in this channel.
- Return type:
- 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)