Channel

The Channel class is the core component for real-time communication in PyPhoenix. It provides Phoenix-compatible channel functionality with support for joining, leaving, message pushing, broadcasting, and middleware processing.

Channel Class

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

Bases: object

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

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

None

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:

None

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:

None

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:

None

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:

None

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:

None

Channel Lifecycle

Channels follow a strict lifecycle pattern:

CLOSED - Initial state, not connected
JOINING - Join request sent, waiting for response
JOINED - Successfully joined, can send/receive messages
LEAVING - Leave request sent, waiting for completion
ERRORED - Error occurred, channel unusable

State transitions are managed automatically, but you can check the current state using the state attribute.

Event Handling

Channels support event-based message handling. You can register handlers for specific events:

channel.on("user_joined", handle_user_joined)
channel.on("new_message", handle_new_message)

Built-in events include:

phx_join - Channel join event
phx_leave - Channel leave event
phx_reply - Reply to a push message
phx_error - Channel error event
phx_close - Channel close event

Custom Events

You can define custom event handlers by implementing methods with the on_ prefix:

class ChatChannel(Channel):
    async def on_message(self, payload, socket):
        # Handle "message" events
        await self.broadcast("new_message", payload)

    async def on_typing(self, payload, socket):
        # Handle "typing" events
        await self.broadcast("user_typing", payload, exclude=socket)

Middleware Support

Channels support middleware for cross-cutting concerns:

from pyphoenix import LoggingMiddleware, AuthMiddleware

channel = Channel("room:lobby")
channel.use_middleware(LoggingMiddleware())
channel.use_middleware(AuthMiddleware(auth_function))

Available middleware includes:

LoggingMiddleware - Request/response logging
AuthMiddleware - Authentication
RateLimitMiddleware - Rate limiting

Broadcasting

Channels can broadcast messages to all joined participants:

# Broadcast to all participants
await channel.broadcast("new_message", {"text": "Hello everyone!"})

# Broadcast to all except sender
await channel.broadcast("typing", {"user": "alice"}, exclude=sender_socket)

Error Handling

Channels provide comprehensive error handling:

try:
    response = await channel.join({"token": "invalid"})
except JoinError as e:
    print(f"Failed to join: {e}")
except PyPhoenixTimeoutError as e:
    print(f"Join timed out: {e}")

Common error types:

JoinError - Join operation failed
ChannelError - General channel error
TimeoutError - Operation timed out