Phoenix Application

The Phoenix class is the main application class that manages channels, routing, and serves as the entry point for PyPhoenix applications.

Phoenix Class

class pyphoenix.phoenix.Phoenix

Bases: object

Main Phoenix application class.

This is the central coordination point for the PyPhoenix application, managing transports, channels, and providing the main API.

__init__()

Initialize the Phoenix application.

__init__()

Initialize the Phoenix application.

channel(topic_pattern)

Decorator to register a channel handler for a topic pattern.

Parameters:

topic_pattern (str) – Topic pattern to handle (e.g., “room:*”)

Returns:

Decorator function

async start(host='localhost', port=4000)

Start the Phoenix application.

Parameters:

• host (str) – Host to bind to

• port (int) – Port to bind to

Return type:

None

async stop()

Stop the Phoenix application.

Return type:

None

async route_message(socket, message)

Route incoming message to appropriate channel.

Parameters:

• socket – The socket that received the message

• message (Message) – The parsed message

Return type:

None

async get_or_create_channel(topic, socket)

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)

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)

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:

None

Channel Registration

The Phoenix application uses decorators to register channel handlers for specific topic patterns:

from pyphoenix import Phoenix, Channel

app = Phoenix()

@app.channel("room:*")
class RoomChannel(Channel):
    async def on_join(self, payload, socket):
        return {"status": "ok"}

@app.channel("user:*")
class UserChannel(Channel):
    async def on_join(self, payload, socket):
        # Only allow users to join their own channel
        user_id = self.topic.split(":")[-1]
        if payload.get("user_id") != user_id:
            return {"status": "error", "response": {"reason": "Unauthorized"}}
        return {"status": "ok"}

Topic Patterns

Topic patterns support wildcards for flexible routing:

• room:* - Matches room:lobby, room:general, etc.

• user:* - Matches user:123, user:alice, etc.

• game:*:* - Matches game:chess:123, game:poker:456, etc.

• exact_topic - Matches only the exact topic name

Application Lifecycle

Phoenix applications have a simple lifecycle:

1. Create - Instantiate the Phoenix class

2. Configure - Register channels and set configuration

3. Start - Begin accepting connections

4. Stop - Gracefully shutdown

import asyncio
from pyphoenix import Phoenix

app = Phoenix()

# Register channels
@app.channel("room:*")
class RoomChannel(Channel):
    pass

# Start the application
async def main():
    try:
        await app.start("localhost", 4000)
    except KeyboardInterrupt:
        await app.stop()

asyncio.run(main())

Configuration

Phoenix applications can be configured using the configuration system:

from pyphoenix import Phoenix, PhoenixConfig, set_config

# Create configuration
config = PhoenixConfig(
    host="0.0.0.0",
    port=4000,
    debug=False
)

# Customize settings
config.transport.max_connections = 10000
config.security.rate_limit_enabled = True

# Apply configuration globally
set_config(config)

app = Phoenix()

Integration with PubSub and Presence

Phoenix applications automatically integrate with the PubSub and Presence systems:

from pyphoenix import get_pubsub, get_presence

@app.channel("room:*")
class RoomChannel(Channel):
    async def on_join(self, payload, socket):
        # Track presence
        presence = get_presence()
        await presence.track(
            pid=payload["user_id"],
            topic=self.topic,
            meta={"username": payload["username"]}
        )

        # Subscribe to updates
        pubsub = get_pubsub()
        await pubsub.subscribe(f"{self.topic}:updates", self._handle_update)

        return {"status": "ok"}

Error Handling

Phoenix applications provide centralized error handling:

@app.on_error
async def handle_error(error, context):
    logger.error("Application error", error=str(error), context=context)

@app.on_channel_error
async def handle_channel_error(channel, error):
    logger.error("Channel error", topic=channel.topic, error=str(error))