# framework

**Repository Path**: wanquan9527/framework

## Basic Information

- **Project Name**: framework
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-25
- **Last Updated**: 2026-03-27

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Framework — 微服务数据流框架

基于 Python 的微服务数据流框架，支持数据生产、处理、消费全链路管理。

## 架构

```
Producer ──MQTT──▶ Processor ──MQTT──▶ Consumer
   │                  │                   │
   └──── Redis Pub/Sub (服务注册发现) ─────┘
                      │
                API Center (RESTful)
```

## 模块说明

| 模块 | 说明 | 入口 |
|------|------|------|
| `common` | 配置、日志、异常、数据模型 | — |
| `registry` | 服务注册中心 (Redis Pub/Sub) | `ServiceRegistry` |
| `queue` | MQTT 消息队列封装 | `MQTTClient` |
| `producer` | 数据生产单元 | `Producer` |
| `processor` | 数据处理器 (Pipeline) | `DataProcessor` |
| `consumer` | 数据消费单元 | `Consumer` |
| `api` | API 管理中心 (FastAPI) | `create_app()` |

## 快速开始

### 1. 安装依赖

```bash
pip install -e ".[dev]"
```

### 2. 启动基础设施

```bash
# Redis
redis-server

# MQTT Broker (Mosquitto)
mosquitto -v
```

### 3. 启动 API 管理中心

```bash
python examples/run_api.py --config configs/api.example.yaml
# 访问 http://localhost:8000/api/docs
```

### 4. 运行数据流演示

```bash
python examples/demo_flow.py
```

## 配置

所有配置支持 YAML 文件 + 环境变量覆盖（前缀 `FW_`）：

```bash
# 环境变量示例
export FW_REDIS__HOST=10.0.0.1
export FW_MQTT__BROKER=10.0.0.2
export FW_LOG_LEVEL=DEBUG
```

配置文件模板见 `configs/` 目录。

## Pipeline 示例

```python
from framework.processor import Pipeline

pipeline = (
    Pipeline("my-pipeline")
    .transform(lambda d: {**d, "value": d["value"] * 2})
    .filter(lambda d: d["value"] > 10)
    .transform(lambda d: {"result": d["value"]})
)
```

## 测试

```bash
pytest tests/ -v --cov=framework
```

## 文档

- [开发计划](docs/PLAN.md)
- [测试标准](docs/TEST_STANDARDS.md)
- [影响范围](docs/IMPACT_SCOPE.md)