Metadata-Version: 2.4
Name: phoenix-channels-python-client
Version: 0.2.5
Summary: A Python client library for connecting to Phoenix Channels
Author: Phoenix Channels Python Client
License-Expression: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: websockets>=14.2
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio>=1.4; extra == "dev"
Requires-Dist: pre-commit>=3.0.0; extra == "dev"
Requires-Dist: commitizen>=3.13.0; extra == "dev"
Requires-Dist: pyrefly; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-asyncio>=1.4; extra == "test"
Dynamic: license-file

# Phoenix Channels Python Client

[![PyPI version](https://img.shields.io/pypi/v/phoenix-channels-python-client.svg)](https://pypi.org/project/phoenix-channels-python-client/)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

A modern, async Python client library for connecting to [Phoenix Channels](https://hexdocs.pm/phoenix/channels.html) - the real-time WebSocket layer of the Phoenix Framework.

## Installation

### For Users

```bash
pip install phoenix-channels-python-client
```

Or from source:

```bash
git clone https://github.com/band-ai/phoenix-channels-python-client.git
cd phoenix-channels-python-client
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install .
```

### For Development

Set up development environment:

```bash
git clone https://github.com/band-ai/phoenix-channels-python-client.git
cd phoenix-channels-python-client
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
python3 -m pip install -U pip
pip install -e ".[dev]"
```

Run tests:

```bash
pytest
```

### Dependencies

- Python 3.11+
- `websockets>=14.2`

## Quick Start

Here's a minimal example to get you started:

```python
import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def main():
    # Connect to Phoenix WebSocket server
    async with PHXChannelsClient(
        websocket_url="ws://localhost:4000/socket/websocket",
        api_key="your-api-key"
    ) as client:
        
        # Define message handler
        async def handle_message(message):
            print(f"Received: {message}")
        
        # Subscribe to a topic
        await client.subscribe_to_topic("room:lobby", handle_message)
        
        # Ctrl+C or SIGTERM shuts the client down and run_forever() returns.
        # The previous handlers are restored first, so a second signal reaches them.
        await client.run_forever()

if __name__ == "__main__":
    asyncio.run(main())
```

### Owning process signals yourself

If your application owns its process signals (a service, desktop app or test
runner), call `client.run_forever(install_signal_handlers=False)`. It then
leaves signal handlers alone; stop the client by scheduling
`client.shutdown(reason)` on the client's event loop, and `run_forever()`
returns `None`:

```python
loop = asyncio.get_running_loop()
async with asyncio.TaskGroup() as tasks:
    loop.add_signal_handler(
        signal.SIGTERM, lambda: tasks.create_task(client.shutdown("SIGTERM"))
    )
    try:
        await client.run_forever(install_signal_handlers=False)
    finally:
        # Once the group exits, a later SIGTERM must not reach the closed group.
        loop.remove_signal_handler(signal.SIGTERM)
```

From a plain `signal.signal` handler or another thread, use
`loop.call_soon_threadsafe(...)` instead: a bare `asyncio.create_task` there
doesn't wake a loop blocked on I/O, so the shutdown can stall until the next
network activity. See [`examples/embedded_host_signals.py`](examples/embedded_host_signals.py).

With the default, handlers are installed only when `run_forever()` runs on the
main thread; elsewhere it just waits. A handler you registered with
`loop.add_signal_handler` before calling `run_forever()` still fires for the
same signal; one registered while it runs replaces the client's.

## Phoenix System Events

Phoenix reserves `phx_join`, `phx_reply`, `phx_leave`, `phx_close` and
`phx_error`. Don't use these names for your own events.

- The client sends `phx_join` and `phx_leave` and consumes their `phx_reply`.
- A server `phx_close` or `phx_error` reaches your message handler like any
  other message; the client does not rejoin or unsubscribe on its own.
- For an event-specific handler on these, register `PHXEvent.close` or
  `PHXEvent.error` from `phoenix_channels_python_client.phx_messages`; the
  plain strings don't match.

## Protocol Versions

Phoenix Channels supports two protocol versions. Choose based on your Phoenix server version:

### Protocol v2.0 (Default)
```python
from phoenix_channels_python_client import PHXChannelsClient

async with PHXChannelsClient(
    websocket_url="ws://localhost:4000/socket/websocket",
    api_key="your-api-key"
    # protocol_version defaults to v2.0
) as client:
    ...  # your code here
```

### Protocol v1.0
```python
from phoenix_channels_python_client import PHXChannelsClient, PhoenixChannelsProtocolVersion

async with PHXChannelsClient(
    websocket_url="ws://localhost:4000/socket/websocket", 
    api_key="your-api-key",
    protocol_version=PhoenixChannelsProtocolVersion.V1
) as client:
    ...  # your code here
```

The client adds `vsn=2.0.0` (V2) or `vsn=1.0.0` (V1) to the socket URL.

## Usage Examples

### Basic Topic Subscription

```python
import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def message_handler(message):
    print(f"Topic: {message.topic}")
    print(f"Event: {message.event}")
    print(f"Payload: {message.payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        await client.subscribe_to_topic("chat:general", message_handler)
        
        # Use built-in method to keep connection alive
        await client.run_forever()

asyncio.run(main())
```

## Message Handlers vs Event-Specific Handlers

The library provides two complementary ways to handle incoming messages:

**Message Handler** - Receives ALL messages for a topic:
- Set when subscribing to a topic or separately with `set_message_handler()`
- Gets the complete message object with topic, event, payload, etc.
- Good for logging, debugging, or handling any message type

**Event-Specific Handlers** - Receive only messages with specific event names:
- Added with `add_event_handler()` for particular events like "user_join", "chat_message", etc.
- Only get the payload (data) part of the message
- Perfect for handling specific application events

**Execution order** when you have both types of handlers:
1. **Message handler runs first** (if set) - receives the full message
2. **Event-specific handler runs second** (if matching event) - receives just the payload

All handlers must be `async def`. You can add, remove, or change either type of handler at any time during your connection.

### Event-Specific Handlers

You can register handlers for specific events on a topic:

```python
import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def handle_user_join(payload):
    print(f"User joined: {payload}")

async def handle_user_leave(payload):
    print(f"User left: {payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        # Subscribe to topic first
        await client.subscribe_to_topic("room:lobby")
        
        # Add event-specific handlers for custom events
        client.add_event_handler("room:lobby", "user_join", handle_user_join)
        client.add_event_handler("room:lobby", "user_leave", handle_user_leave)
        
        # Use built-in method to keep connection alive
        await client.run_forever()

asyncio.run(main())
```

## Unsubscribing from Topics

You can unsubscribe from topics to stop receiving messages and clean up resources:

```python
import asyncio
from phoenix_channels_python_client import PHXChannelsClient

async def on_message(msg):
    print(f"Message: {msg.payload}")

async def main():
    async with PHXChannelsClient("ws://localhost:4000/socket/websocket", "api-key") as client:
        # Subscribe to a topic
        await client.subscribe_to_topic("room:lobby", on_message)
        
        # Do some work...
        await asyncio.sleep(5)
        
        # Unsubscribe when no longer needed
        await client.unsubscribe_from_topic("room:lobby")
        print("Unsubscribed from room:lobby")
        
        # Continue with other work or subscribe to different topics
        await client.run_forever()

asyncio.run(main())
```

**Notes on unsubscription:**
- Removes all handlers (both message and event-specific) for that topic
- Sends `phx_leave` and waits for the reply (`leave_timeout_s`, default 5 s)
- Raises `PHXTopicError` if the topic isn't subscribed or the leave times out, and `PHXConnectionError` while disconnected

## Configuration

Everything after `api_key` is keyword-only:

| Option | Default | Meaning |
|---|---|---|
| `protocol_version` | `V2` | Phoenix Channels protocol version |
| `auto_reconnect` | `True` | Reconnect and rejoin topics after a disconnect |
| `reconnect_policy` | `ReconnectPolicy()` | Backoff and close-code handling (see below) |
| `heartbeat_interval_s` | `30.0` | Heartbeat period; `None` disables heartbeats |
| `join_timeout_s` | `10.0` | Wait for a join reply |
| `leave_timeout_s` | `5.0` | Wait for a leave reply |
| `max_topic_queue_size` | `1000` | Per-topic buffer; the oldest message is dropped when full |
| `callback_drain_timeout_s` | `2.0` | How long a running callback may finish before a rejoin cancels it |
| `on_reconnect` | `None` | `async () -> None`, called after topics are rejoined |
| `on_disconnect` | `None` | `async (error: Exception \| None) -> None`; `None` on a clean close |
| `on_heartbeat_ack` | `None` | Synchronous `() -> None`; runs on the message path, so keep it fast |
| `additional_headers` | `None` | Extra handshake headers sent on every (re)connect |

Exceptions raised in callbacks are logged, not raised.

## Connection Lifecycle and Errors

- `async with` waits for the first connection. With `auto_reconnect=True` it keeps retrying; with `auto_reconnect=False` a failed first connection raises `PHXConnectionError`.
- `run_forever()` returns `None` after `shutdown()`, a signal, or a close that doesn't reconnect. It raises `PHXConnectionError` on a terminal close, after repeated rapid disconnects, or when the client has never entered `async with`.
- `subscribe_to_topic()` raises `PHXTopicError` on a rejected join, a join timeout or a duplicate subscription.
- `await client.close_connection(reason)` force-closes the current connection; the client then reconnects if `auto_reconnect` is on.
- Exceptions live in `phoenix_channels_python_client.exceptions`.

### Reconnects

| Close code | Behaviour (defaults) |
|---|---|
| 1000, 1001 | Stop, unless `reconnect_on_normal_close=True` |
| 1008 | Stop with `PHXConnectionError` (`policy_violation_is_terminal=True`) |
| 1012 | Reconnect after at least a random 1–5 s |
| 1013 | Reconnect after at least a random 30–60 s |
| Anything else, or no close frame | Reconnect with jittered backoff: 0.5 s × 2ⁿ, capped at 30 s, reset after 60 s of uptime |

Disconnects within 5 s of connecting count as rapid and get longer minimum delays; ten within 60 s stop the client with `PHXConnectionError`. Tune all of this with `ReconnectPolicy`.

### Heartbeats

The client sends a `heartbeat` on the `phoenix` topic every `heartbeat_interval_s`. A missed reply only logs a warning; to act on it, combine `on_heartbeat_ack` with `close_connection()`.

## Logging

```python
import logging
from phoenix_channels_python_client import setup_logging

setup_logging(logging.DEBUG)  # loggers live under "phoenix_channels_python_client"
```

## Examples

- [`examples/minimal_phx_events_client.py`](examples/minimal_phx_events_client.py): subscribe to a topic and log messages until Ctrl+C or SIGTERM.
- [`examples/embedded_host_signals.py`](examples/embedded_host_signals.py): run the client in a host that owns its process signals.

---

## Security Considerations

**API Key in URL**: This client passes the API key as a URL query parameter when connecting to the WebSocket (e.g., `wss://server.com/socket?api_key=xxx`). Even over WSS (encrypted in transit), the API key may be logged by:

- Server access logs
- Reverse proxies and load balancers
- Network monitoring tools

**Recommendations:**
- Ensure your infrastructure does not log full URLs in production
- Use short-lived, rotatable tokens rather than long-lived API keys
- Consider this limitation when evaluating this client for sensitive environments

**Note:** The official Phoenix JS client (v1.8+) supports header-based authentication via the `authToken` option, which avoids URL logging. This client always sends `api_key` in the URL (logged URLs show `api_key=***`). Pass `additional_headers={"x-api-key": key}` to also send it as a handshake header on every (re)connect, e.g. for a proxy that reads it from there.

---

**Need help?** Open an issue on GitHub or check the [Phoenix Channels documentation](https://hexdocs.pm/phoenix/channels.html) for more information about the Phoenix Channels protocol.
