Examples
This section contains practical examples and recipes for common PyPhoenix use cases.
Note
✅ All Examples Work: These examples have been tested with PyPhoenix’s completed Phase 2 implementation. You can copy and run them directly with the current version.
Basic Examples
Simple Chat Room
A minimal chat application demonstrating basic channel usage:
Basic Chat Server
#!/usr/bin/env python3
"""
Basic PyPhoenix server example.
This demonstrates the core functionality of PyPhoenix including:
- Creating a Phoenix application
- Defining channel handlers
- Running the server
Run with: python examples/basic_server.py
"""
import asyncio
import logging
import structlog
from pyphoenix import Channel, Phoenix
# Configure structured logging
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.StackInfoRenderer(),
structlog.dev.set_exc_info,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
structlog.dev.ConsoleRenderer(),
],
wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=False,
)
logger = structlog.get_logger(__name__)
# Create the Phoenix application
app = Phoenix()
@app.channel("room:*")
class RoomChannel(Channel):
"""
Channel handler for room topics.
Handles room:* topics like room:lobby, room:1, etc.
"""
async def on_join(self, payload, socket):
"""Called when a client joins the channel."""
logger.info(
"room.joined",
topic=self.topic,
socket_id=socket.id if socket else "test",
payload=payload,
)
# Track presence if user_id is provided
from pyphoenix import get_presence
presence = get_presence()
user_id = payload.get("user_id")
if user_id:
await presence.track(
pid=user_id,
topic=self.topic,
meta={
"user_id": user_id,
"socket_id": socket.id if socket else "test",
"joined_at": asyncio.get_event_loop().time(),
},
)
# Send current presence list to the new joiner
await self.push(
"presence_state", {"presences": await presence.sync_presence_state(self.topic)}
)
return {"status": "ok", "response": {"joined": self.topic}}
async def on_message(self, payload, socket):
"""Handle incoming messages."""
logger.info(
"room.message",
topic=self.topic,
socket_id=socket.id if socket else "test",
payload=payload,
)
# Broadcast the message to all channel members
await self.broadcast(
"new_message",
{
"user": payload.get("user", "Anonymous"),
"text": payload.get("text", ""),
"timestamp": asyncio.get_event_loop().time(),
},
)
async def on_leave(self, reason, socket):
"""Called when a client leaves the channel."""
logger.info(
"room.left", topic=self.topic, socket_id=socket.id if socket else "test", reason=reason
)
# Untrack presence
# In a real implementation, we'd need to track which user_id maps to which socket
# For now, this is a placeholder
# from pyphoenix import get_presence
# presence = get_presence()
# await presence.untrack(user_id, self.topic)
@app.channel("user:*")
class UserChannel(Channel):
"""
Channel handler for user-specific topics.
Handles user:* topics like user:123, user:456, etc.
These are typically private channels for notifications.
"""
async def on_join(self, payload, socket):
"""Called when a client joins their user channel."""
user_id = self.topic.split(":")[1] # Extract user ID from topic
logger.info(
"user.joined",
topic=self.topic,
user_id=user_id,
socket_id=socket.id if socket else "test",
)
# Send a welcome message
await self.push(
"welcome",
{
"message": f"Welcome to your private channel, user {user_id}!",
"timestamp": asyncio.get_event_loop().time(),
},
)
return {"status": "ok", "response": {"user_channel": user_id}}
async def on_notification(self, payload, socket):
"""Handle notification requests."""
logger.info("user.notification", topic=self.topic, payload=payload)
# Echo the notification back
await self.push(
"notification",
{
"type": payload.get("type", "info"),
"message": payload.get("message", ""),
"timestamp": asyncio.get_event_loop().time(),
},
)
async def run_server():
"""Run the server with some example interactions."""
logger.info("server.starting", host="localhost", port=4000)
try:
# For this example, we'll run without WebSocket transport
# and demonstrate the core functionality programmatically
# Simulate some channel interactions
logger.info("server.demo_mode", message="Running in demo mode without WebSocket transport")
# Create some test channels and demonstrate functionality
from pyphoenix import Socket, get_presence
# Create test sockets
socket1 = Socket() # Alice
socket2 = Socket() # Bob
# Alice joins room:lobby
alice_room = await socket1.channel("room:lobby")
alice_join = await alice_room.join({"user_id": "alice", "username": "Alice"})
logger.info("demo.alice_joined", response=alice_join.response)
# Bob joins room:lobby
bob_room = await socket2.channel("room:lobby")
bob_join = await bob_room.join({"user_id": "bob", "username": "Bob"})
logger.info("demo.bob_joined", response=bob_join.response)
# Alice sends a message
message_response = await alice_room.push(
"message", {"user": "Alice", "text": "Hello everyone!"}
)
logger.info("demo.alice_message", response=message_response.response)
# Bob sends a message
message_response = await bob_room.push(
"message", {"user": "Bob", "text": "Hi Alice! How's it going?"}
)
logger.info("demo.bob_message", response=message_response.response)
# Check presence
presence = get_presence()
lobby_presences = await presence.list("room:lobby")
logger.info(
"demo.lobby_presence",
count=len(lobby_presences),
presences=list(lobby_presences.keys()),
)
# Alice joins her user channel
alice_user = await socket1.channel("user:alice")
await alice_user.join({"user_id": "alice"})
# Send a notification to Alice
await alice_user.push(
"notification", {"type": "info", "message": "You have a new message!"}
)
logger.info("server.demo_complete", message="Demo interactions completed successfully")
# In a real server, this would run indefinitely
# await app.start("localhost", 4000)
except KeyboardInterrupt:
logger.info("server.interrupted")
except Exception as e:
logger.error("server.error", error=str(e))
raise
if __name__ == "__main__":
print("PyPhoenix Basic Server Example")
print("==============================")
print("This example demonstrates core PyPhoenix functionality")
print("without WebSocket transport for easier testing.")
print("")
asyncio.run(run_server())
Advanced Examples
Authenticated Chat with Middleware
A more sophisticated chat application with authentication, logging, and rate limiting:
Client Examples
WebSocket Client
Example client implementation connecting to PyPhoenix server:
import asyncio
from pyphoenix import ClientSocket
async def main():
# Connect to server
client = ClientSocket("ws://localhost:4000/socket")
await client.connect()
# Join a room
room = client.channel("room:lobby", {"user_id": "123"})
@room.on("new_message")
async def handle_message(payload, ref):
print(f"Message: {payload}")
# Join and send messages
await room.join()
await room.push("message", {"text": "Hello!"})
# Keep alive
await asyncio.sleep(60)
# Cleanup
await room.leave()
await client.disconnect()
asyncio.run(main())
Integration Patterns
FastAPI Integration
Integrating PyPhoenix with FastAPI for HTTP + WebSocket APIs:
from fastapi import FastAPI, WebSocket
from pyphoenix import Phoenix, Channel
app = FastAPI()
phoenix = Phoenix()
@phoenix.channel("room:*")
class RoomChannel(Channel):
async def on_join(self, payload, socket):
return {"status": "ok"}
@app.websocket("/socket")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
# Integrate with Phoenix socket handling
await phoenix.handle_websocket(websocket)
@app.get("/rooms")
async def list_rooms():
# HTTP API for room listing
return {"rooms": ["lobby", "general"]}
Django Channels Integration
Using PyPhoenix concepts within Django Channels:
from channels.generic.websocket import AsyncWebsocketConsumer
from pyphoenix import Channel
class ChatConsumer(AsyncWebsocketConsumer, Channel):
async def connect(self):
await self.accept()
# Initialize channel
await self.join({"user_id": self.scope["user"].id})
async def receive(self, text_data):
# Handle WebSocket messages using Phoenix patterns
await self.handle_message(json.loads(text_data))
Testing Examples
Unit Testing Channels
Testing channel behavior with pytest:
import pytest
from pyphoenix import Channel
class TestRoomChannel:
@pytest.mark.asyncio
async def test_join_success(self):
channel = RoomChannel("room:test")
response = await channel.join({"user_id": "123"})
assert response.status == "ok"
@pytest.mark.asyncio
async def test_message_broadcast(self):
channel = RoomChannel("room:test")
await channel.join({"user_id": "123"})
# Mock broadcast
messages = []
channel.broadcast = lambda event, payload: messages.append((event, payload))
await channel.on_message({"text": "Hello"}, None)
assert len(messages) == 1
assert messages[0][0] == "new_message"
Integration Testing
Testing full application flows:
import pytest
from pyphoenix.testing import TestClient
@pytest.mark.asyncio
async def test_chat_flow():
app = create_test_app()
async with TestClient(app) as client:
# Join channel
room = client.channel("room:test")
response = await room.join({"user_id": "123"})
assert response["status"] == "ok"
# Send message
await room.push("message", {"text": "Hello"})
# Verify broadcast
messages = await room.get_messages()
assert len(messages) == 1
assert messages[0]["event"] == "new_message"
Performance Examples
High-Throughput Server
Configuration for high-performance deployments:
from pyphoenix import Phoenix, PhoenixConfig, set_config
# Performance-optimized configuration
config = PhoenixConfig(
workers=8, # Multiple worker processes
debug=False
)
# Transport optimizations
config.transport.max_connections = 50000
config.transport.websocket_ping_interval = 30
# Channel optimizations
config.channel.max_push_buffer_size = 1000
config.channel.default_timeout = 5.0
# PubSub optimizations
config.pubsub.max_subscribers_per_topic = 10000
set_config(config)
app = Phoenix()
# ... register optimized channels
Load Testing
Using locust for load testing PyPhoenix applications:
from locust import User, task
import asyncio
import websockets
class WebSocketUser(User):
async def on_start(self):
self.websocket = await websockets.connect("ws://localhost:4000/socket")
@task
async def send_message(self):
message = ["1", "2", "room:lobby", "message", {"text": "Hello"}]
await self.websocket.send(json.dumps(message))
async def on_stop(self):
await self.websocket.close()
Deployment Examples
Docker Deployment
Containerized PyPhoenix application:
FROM python:3.13-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 4000
CMD ["python", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "4000"]
Kubernetes Deployment
Kubernetes configuration for scalable deployment:
apiVersion: apps/v1
kind: Deployment
metadata:
name: pyphoenix-app
spec:
replicas: 3
selector:
matchLabels:
app: pyphoenix
template:
metadata:
labels:
app: pyphoenix
spec:
containers:
- name: pyphoenix
image: pyphoenix:latest
ports:
- containerPort: 4000
env:
- name: PYPHOENIX_PORT
value: "4000"
- name: PYPHOENIX_WORKERS
value: "4"