Metadata-Version: 2.4
Name: agent-redis-framework
Version: 1.1.2
Summary: A clean, elegant Python library for Redis-based multi-task scheduling with SortedSets and Streams.
Requires-Python: >=3.12
Requires-Dist: redis<7.0,>=5.0
Requires-Dist: typing-extensions<5.0,>=4.12
Provides-Extra: dev
Requires-Dist: fastapi>=0.104.0; extra == 'dev'
Requires-Dist: pydantic>=2.0.0; extra == 'dev'
Requires-Dist: uvicorn[standard]>=0.24.0; extra == 'dev'
Description-Content-Type: text/markdown

# Agent Redis Framework

一个优雅、高效的 Python Redis 框架，专为多任务调度与消息流处理而设计。提供基于 Redis Sorted Sets 的轻量任务队列和基于 Redis Streams 的消费组封装，适合构建可扩展的分布式任务系统。

## ✨ 核心特性

### 🚀 SortedSetQueue - 智能任务队列
- **优先级调度**: 基于 Redis Sorted Set 实现的轻量任务队列
- **原子操作**: 使用 `ZPOPMIN`/`ZPOPMAX` 确保多消费者场景下的任务安全分发
- **灵活排序**: 支持按分数升序/降序弹出任务
- **失败处理**: 内置任务处理失败的回调机制
- **批量处理**: 支持一次性弹出并处理多个任务

### 📡 RedisStreamsClient - 流式消息处理
- **消费组管理**: 完整的 Redis Streams 消费组封装
- **自动 ACK**: 消息处理成功后自动确认
- **流量控制**: 支持流长度限制和自动修剪
- **阻塞消费**: 可配置的阻塞时间和批量消费
- **错误恢复**: 处理待确认消息和消费者故障恢复

### 🔧 企业级特性
- **连接池管理**: 自动连接池复用，优化高并发性能
- **环境配置**: 灵活的 `.env` 文件和环境变量支持
- **类型安全**: 完整的 Python 类型注解
- **线程安全**: 所有核心组件支持多线程并发
- **零依赖**: 仅依赖 `redis` 和 `typing-extensions`

## 🚀 快速开始

### Redis 连接配置

#### 环境变量配置

创建 `.env` 文件：
```bash
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=your_password
REDIS_MAX_CONNECTIONS=20
```

```python
from agent_redis_framework import get_redis

# 自动从环境变量加载配置
redis_client = get_redis()
print(redis_client.ping())  # True
```

### SortedSetQueue - 任务队列示例

```python
from agent_redis_framework.sortedset.sorted_queue import SortedSetQueue, Task
import time

queue = SortedSetQueue()
queue_key = "my_task_queue"

# 清空队列（可选）
queue.clear(queue_key)

# 推送任务（分数越小优先级越高）
import json
queue.push(queue_key, Task(id="urgent_task", payload=json.dumps({"priority": "high"})), score=1)
queue.push(queue_key, Task(id="normal_task", payload=json.dumps({"priority": "normal"})), score=5)
queue.push(queue_key, Task(id="delayed_task", payload=json.dumps({"priority": "low"})), score=10)

# 定义任务处理函数
def process_task(score: float, task: Task) -> bool:
    print(f"Processing task {task.id} at score {score}: {task.payload}")
    # 模拟任务处理
    time.sleep(0.1)
    return True  # 返回 True 表示处理成功

# 定义失败处理函数
def handle_failure(score: float, task: Task) -> None:
    print(f"Task {task.id} failed at score {score}, logging for retry...")

# 原子弹出并处理任务
processed = queue.pop_and_handle(
    queue_key,
    callback=process_task,
    on_failure=handle_failure,
    count=2  # 一次处理 2 个任务
)

print(f"Processed {len(processed)} tasks")
print(f"Remaining tasks: {queue.size(queue_key)}")
```

### RedisStreamsClient - 流式消息示例

- StreamMsg.meta 的类型已收紧为：`dict[str, bytes | str | int | float]`
- push() 会将 meta 扁平化写入 Redis，字段名前缀为 `__m_`（例如：`__m_trace_id`）
- consume() 的回调签名已更新为：`callback(msg_key: str, msg: StreamMsg) -> bool`

```python
from agent_redis_framework import RedisStreamsClient, StreamMsg, get_redis
import threading
import time
import json

# 初始化
redis_client = get_redis()
stream_client = RedisStreamsClient(redis_client=redis_client)

stream_name = "user_events"
group_name = "analytics_group"
consumer_name = "consumer_1"

# 创建消费组（幂等操作）
stream_client.ensure_group(stream_name, group_name)

# 推送消息（payload 建议为 JSON 字符串；meta 仅允许标量值）
msg = StreamMsg(
    stream=stream_name,
    payload=json.dumps({
        "event_type": "user_login",
        "user_id": "12345",
        "timestamp": time.time(),
        "ip_address": "192.168.1.100"
    }),
    meta={"source": "readme", "request_id": "req-1", "retry": 0}
)
msg_key = stream_client.push(msg)
print(f"Message pushed with ID: {msg_key}")

```

# 定义消息处理函数（新版签名）
def handle_message(msg_key: str, msg: StreamMsg) -> bool:
    print(f"Processing message {msg_key} from {msg.stream}")
    print(f"Payload: {msg.payload}")  # 若为 JSON 字符串，可自行 json.loads
    print(f"Meta: {msg.meta}")       # 来自 __m_ 前缀字段
    time.sleep(0.1)
    return True

# 在单独线程中启动消费者
def start_consumer():
    stream_client.consume(
        streams=[stream_name],
        group=group_name,
        consumer=consumer_name,
        callback=handle_message,   # 注意：签名为 (msg_key, msg)
        block_ms=5000,             # 5秒阻塞超时
        count=10                   # 每次最多读取10条消息
    )

consumer_thread = threading.Thread(target=start_consumer, daemon=True)
consumer_thread.start()

# 继续推送更多消息
for i in range(3):
    msg = StreamMsg(
        stream=stream_name,
        payload=json.dumps({
            "event_type": "page_view",
            "user_id": f"user_{i}",
            "page": f"/page/{i}",
            "timestamp": time.time()
        }),
        meta={"source": "readme", "seq": i}
    )
    stream_client.push(msg)
    time.sleep(1)
```

## 🏗️ 高级用法

### 流消息批量处理

```python
from agent_redis_framework import RedisStreamsClient, StreamMsg, get_redis
import json

stream_client = RedisStreamsClient(redis_client=get_redis())

# 批量推送消息
messages = [
    {"order_id": f"order_{i}", "amount": i * 100, "status": "pending"}
    for i in range(100)
]

for msg in messages:
    msg = StreamMsg(stream="order_stream", payload=json.dumps(msg), meta={"batch": 1})
    stream_client.push(msg, maxlen=1000)  # 限制流长度

# 批量消费处理
def batch_process_orders(msg_key: str, msg: StreamMsg) -> bool:
    order_data = json.loads(msg.payload) if msg.payload else {}
    print(f"Processing order: {order_data.get('order_id', 'unknown')} with key {msg_key}")
    return True

stream_client.consume(
    streams=["order_stream"],
    group="order_processors",
    consumer="processor_1",
    callback=batch_process_orders,
    count=50,  # 每次批量处理50条消息
    block_ms=1000
)
```

## 📚 API 参考（关键变更）

### RedisStreamsClient
- `push(msg, maxlen=None)` - 将 StreamMsg 推送到流中；meta 会以 `__m_` 前缀扁平化写入
- `consume(streams, group, consumer, callback, block_ms=5000, count=1, read_new_only=True)` - 消费消息；`callback` 签名现为 `(msg_key: str, msg: StreamMsg) -> bool`

### 数据类
- `StreamMsg(stream, payload, meta={})` - 流消息数据类，payload 为字符串格式；meta 类型为 `dict[str, bytes | str | int | float]`

注意：复杂结构（如列表/字典）请先 JSON 序列化后放入 payload；meta 仅接受标量类型。如果一定要在 meta 保留复杂结构，请先自行序列化为字符串。