Metadata-Version: 2.4
Name: n1r-phoenix
Version: 5.0.0
Summary: N1 Research's Phoenix Platform - Resilient Redis Streams-based worker framework with decorator API
Project-URL: Homepage, https://github.com/n1healthcare/n1r-phoenix
Project-URL: Repository, https://github.com/n1healthcare/n1r-phoenix
Project-URL: Documentation, https://n1healthcare.github.io/n1r-phoenix
Project-URL: Issues, https://github.com/n1healthcare/n1r-phoenix/issues
Author-email: Arun <arun@n1.care>
License: Proprietary
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.12
Requires-Dist: httpx>=0.28.1
Requires-Dist: opentelemetry-api>=1.39.1
Requires-Dist: opentelemetry-exporter-otlp>=1.39.1
Requires-Dist: opentelemetry-sdk>=1.39.1
Requires-Dist: pydantic-settings>=2.13.1
Requires-Dist: pydantic>=2.12.5
Requires-Dist: redis[hiredis]<7.2,>=7.1.0
Requires-Dist: structlog>=25.5.0
Provides-Extra: dev
Requires-Dist: black<26,>=25.12.0; extra == 'dev'
Requires-Dist: fakeredis>=2.34.1; extra == 'dev'
Requires-Dist: hatch>=1.16.5; extra == 'dev'
Requires-Dist: lupa>=2.6; extra == 'dev'
Requires-Dist: mypy>=1.19.1; extra == 'dev'
Requires-Dist: pytest-asyncio>=1.3.0; extra == 'dev'
Requires-Dist: pytest-cov>=7.0.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.36.0; extra == 'dev'
Requires-Dist: pytest>=9.0.2; extra == 'dev'
Requires-Dist: ruff>=0.15.4; extra == 'dev'
Description-Content-Type: text/markdown

# N1R Phoenix

A resilient, Redis Streams-based worker framework for Python.

Phoenix handles the hard parts of building distributed workers: locking, retries, dead letter queues, and graceful shutdown. You write your business logic; Phoenix handles the infrastructure.

## Installation

```bash
pip install n1r-phoenix
```

Requires Python 3.12+ and Redis 7+.

## Quick Start

Create a worker in 3 lines:

```python
import asyncio
from n1r.phoenix import phoenix_worker, WorkerContext

@phoenix_worker("my-service")
async def process(ctx: WorkerContext, data: dict) -> dict | None:
    print(f"Processing {ctx.record_id} for user {ctx.user_id}")
    # Your business logic here
    return {"status": "done"}

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

Run it:

```bash
export REDIS_URL=redis://localhost:6379/0
python worker.py
```

Send a message:

```python
import asyncio
from n1r.phoenix import QueueManager

async def send():
    qm = QueueManager("redis://localhost:6379/0")
    await qm.connect("producer")

    producer = qm.get_producer()
    await producer.send("my-service-queue", {
        "record_id": "doc-123",
        "user_id": "user-456",
    })

    await qm.close()

asyncio.run(send())
```

## What Phoenix Handles

| Problem | Solution |
|---------|----------|
| Two workers processing the same record | Distributed locking |
| Worker crashes mid-processing | Stale message reclamation |
| Permanent failures blocking the queue | Dead Letter Queue (DLQ) |
| Rate limits from external APIs | Exponential backoff with jitter |
| Cascading failures when APIs are down | Circuit breakers |
| Long jobs timing out | Heartbeats to extend locks |
| Resuming after crashes | Checkpointing |
| Duplicate processing on retries | Idempotency markers |

## Core Concepts

### The `@phoenix_worker` Decorator

The decorator turns your async function into a full worker with `.run()`, `.stop()`, and `.get_worker()` methods:

```python
@phoenix_worker("my-service")
async def process(ctx: WorkerContext, data: dict) -> dict | None:
    # ctx provides: record_id, user_id, message_id, retry_count
    # data is the message payload from Redis
    return {"result": "value"}  # Sent to output_queue if configured
```

### WorkerContext

Every message handler receives a `WorkerContext` with useful information:

```python
async def process(ctx: WorkerContext, data: dict):
    ctx.record_id      # Extracted from data (configurable field)
    ctx.user_id        # Extracted from data (configurable field)
    ctx.message_id     # Redis message ID
    ctx.retry_count    # How many times this message was delivered (0 = first try)
    ctx.redis          # Redis client for cache/checkpoint operations

    # For long-running jobs, extend the lock:
    await ctx.heartbeat()

    # For batch processing, collect related messages:
    additional = await ctx.collect_pending("user_id", max_count=20)
```

### Error Categories

Phoenix classifies errors to decide what to do with failed messages:

| Category | Behavior | Examples |
|----------|----------|----------|
| `RETRYABLE` | Retry with backoff until max_retries, then DLQ | Network errors, 500 responses |
| `TRANSIENT` | Retry forever with longer backoff | Rate limits (429) |
| `NON_RETRYABLE` | Send to DLQ immediately | Validation errors, 404 |
| `FATAL` | Send to DLQ immediately | Critical system errors |

```python
from n1r.phoenix import ErrorRegistry, ErrorCategory

registry = ErrorRegistry()
registry.register_error(MyCustomError, category=ErrorCategory.NON_RETRYABLE)

@phoenix_worker("my-service", error_registry=registry)
async def process(ctx, data):
    ...
```

### Lifecycle Hooks

Initialize resources after Redis connects, clean up before shutdown:

```python
from n1r.phoenix import phoenix_worker, StartupContext, Cache

cache: Cache = None

async def setup(ctx: StartupContext):
    global cache
    cache = Cache(ctx.redis, prefix="my-service")

async def teardown(ctx: StartupContext):
    pass  # Cleanup if needed

@phoenix_worker("my-service", on_startup=setup, on_shutdown=teardown)
async def process(ctx, data):
    cached = await cache.get(f"user:{ctx.user_id}")
    ...
```

## Configuration

Configure via environment variables or `.env` file:

```ini
# Required
SERVICE_NAME=my-service

# Redis
REDIS_URL=redis://localhost:6379/0

# Queue names (defaults derived from SERVICE_NAME)
INPUT_QUEUE=my-service-queue
OUTPUT_QUEUE=next-service-queue
DLQ_NAME=my-service-dlq

# Reliability
MAX_RETRIES=3
LOCK_TIMEOUT=600
LOCK_FIELD=record_id

# Backoff (seconds)
RETRY_INITIAL_BACKOFF=1
RETRY_MAX_BACKOFF=60
TRANSIENT_RETRY_INITIAL_BACKOFF=5
TRANSIENT_RETRY_MAX_BACKOFF=300

# Telemetry (disabled by default)
ENABLE_TRACING=false
ENABLE_METRICS=false
```

## Features

### Caching

```python
from n1r.phoenix import Cache

cache = Cache(ctx.redis, prefix="my-service")

# Set with TTL
await cache.set("key", {"data": "value"}, ttl=3600)

# Get (optionally reconstruct into Pydantic model)
data = await cache.get("key", model=MyModel)

# Batch operations
results = await cache.get_many(["key1", "key2"])
await cache.set_many({"key1": val1, "key2": val2}, ttl=3600)
```

### Checkpointing

Resume long-running jobs after crashes:

```python
from n1r.phoenix import Checkpoints

ckpt = Checkpoints(ctx.redis, scope=ctx.record_id)

# Load previous progress
done = await ckpt.load("page:*", model=PageResult)

for idx, page in enumerate(pages):
    if f"page:{idx}" in done:
        continue  # Skip already processed

    result = await process_page(page)
    await ckpt.save(f"page:{idx}", result)

# Clean up on success
await ckpt.clear()
```

### Idempotency

Prevent duplicate processing:

```python
from n1r.phoenix import IdempotencyManager

idempotency = IdempotencyManager(ctx.redis, "my-service")

if await idempotency.is_processed(ctx.record_id):
    return {"status": "already_processed"}

result = await do_work(data)
await idempotency.mark_processed(ctx.record_id)
```

### Circuit Breaker

Fail fast when external APIs are down:

```python
from n1r.phoenix import CircuitBreaker

cb = CircuitBreaker("openai-api", failure_threshold=5, reset_timeout=60)

async def call_llm():
    return await cb.call(client.chat.completions.create, messages=[...])
```

### Retry with Backoff

Automatic retries for HTTP calls:

```python
from n1r.phoenix import retry_with_backoff, BackoffConfig

@retry_with_backoff(
    max_retries=3,
    backoff_config=BackoffConfig(initial_delay=1.0, max_delay=30.0),
)
async def fetch_data(url: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        return response.json()
```

### Billing Integration

Per-user API keys and spend tracking:

```python
from n1r.phoenix import APIKeyManager, BillingClientConfig

config = BillingClientConfig(
    billing_service_url="http://billing:8000",
    billing_api_key="system-key",
    service_name="my-service",
)
manager = APIKeyManager(ctx.redis, config)

api_key = await manager.get_key(ctx.user_id)
```

## Documentation

- [Getting Started](docs/guide/getting-started.md) - Build your first worker
- [Configuration](docs/guide/configuration.md) - All environment variables
- [Error Handling](docs/guide/error-handling.md) - Retry, DLQ, and error categories
- [Caching](docs/guide/caching.md) - Redis caching patterns
- [Checkpointing](docs/guide/checkpointing.md) - Resume long-running jobs
- [Idempotency](docs/guide/idempotency.md) - Prevent duplicate processing
- [Reliability](docs/guide/reliability.md) - Locks, circuit breakers, backoff
- [Billing](docs/guide/billing.md) - Per-user API keys

## License

Proprietary. Copyright N1 Research LLC.
