# data_weave

**Repository Path**: SSheamus_NO/data_weave

## Basic Information

- **Project Name**: data_weave
- **Description**: 数据编排
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-04-23
- **Last Updated**: 2026-05-06

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 服务编排引擎

支持 HTTP、RPC、Python 代码节点的可视化编排引擎。

## 特性

- **多节点类型**: HTTP、RPC、Python、条件分支、循环、子流程
- **可视化编排**: 拖拽式流程设计器
- **安全执行**: Python 代码沙箱隔离
- **变量插值**: 支持 Jinja2 语法
- **重试机制**: 可配置的重试策略
- **监控告警**: Prometheus 指标暴露
- **Docker 支持**: 一键部署

## 快速开始

### 安装依赖

```bash
pip install -r requirements.txt
```

### 启动服务

```bash
python src/main.py
```

或使用 Docker:

```bash
docker-compose up -d
```

服务将在 http://localhost:8000 启动

### 可视化界面

打开 http://localhost:8000 进入编排界面

API 文档: http://localhost:8000/docs

## 架构

```
┌─────────────────────────────────────────────────────────────┐
│                      编排引擎核心                            │
├─────────────────────────────────────────────────────────────┤
│  FlowParser │ DAGScheduler │ ExecutionContext │ Engine     │
├─────────────────────────────────────────────────────────────┤
│                      节点执行器                              │
├──────────┬──────────┬──────────┬──────────┬────────────────┤
│ HTTP节点 │ RPC节点  │ Python节点│ 条件节点 │  循环节点      │
└──────────┴──────────┴──────────┴──────────┴────────────────┘
```

## 节点类型

### HTTP 节点

```json
{
  "id": "http_1",
  "type": "http",
  "name": "获取用户",
  "config": {
    "method": "GET",
    "url": "http://api.example.com/users/${userId}",
    "timeout": 30
  }
}
```

### Python 节点

```json
{
  "id": "python_1",
  "type": "python",
  "name": "数据处理",
  "config": {
    "code": "def handler(params, context):\n    data = params['http_1']\n    return {'processed': data['name'].upper()}"
  }
}
```

### RPC 节点

```json
{
  "id": "rpc_1",
  "type": "rpc",
  "name": "调用gRPC",
  "config": {
    "service_type": "grpc",
    "service_name": "user-service",
    "method": "GetUser"
  }
}
```

### 条件节点

```json
{
  "id": "cond_1",
  "type": "condition",
  "name": "条件判断",
  "config": {
    "expression": "{{ python_1.processed }} == true",
    "branches": {
      "true": ["http_2"],
      "false": ["http_3"]
    }
  }
}
```

### 循环节点

```json
{
  "id": "loop_1",
  "type": "loop",
  "name": "批量处理",
  "config": {
    "type": "for",
    "collection": "{{ http_1.items }}",
    "max_iterations": 1000
  }
}
```

## API 文档

### 创建流程

```bash
POST /api/v1/flows
{
  "id": "user-flow",
  "name": "用户流程",
  "nodes": [...]
}
```

### 触发执行

```bash
POST /api/v1/executions
{
  "flow_id": "user-flow",
  "input": {"userId": "123"}
}
```

### 查询执行结果

```bash
GET /api/v1/executions/{execution_id}
```

### Prometheus 指标

```bash
GET /metrics
```

## Python 代码节点

### 内置方法

```python
def handler(params, context):
    # 获取上游节点输出
    http_result = params.get("http_node_id")

    # 调用HTTP服务
    api_result = context["call_http"]("GET", "/api/test")

    # 输出日志
    context["log"]("info", "Processing...")

    # 返回结果
    return {"status": "success", "data": api_result}
```

### 安全限制

- 禁止文件操作 (open)
- 禁止动态代码执行 (exec, eval)
- 禁止导入危险模块 (os, subprocess, sys)
- 禁止类自省 (__class__, __bases__)
- 执行超时限制 (默认 30s)
- 内存限制 (128MB)

## 监控指标

| 指标 | 说明 |
|------|------|
| flow_execution_total | 流程执行总数 |
| flow_execution_duration_seconds | 流程执行耗时 |
| node_execution_total | 节点执行总数 |
| python_execution_timeout_total | Python执行超时数 |
| http_request_total | HTTP请求总数 |

## 配置

### 环境变量

| 变量 | 默认值 | 说明 |
|------|--------|------|
| HOST | 0.0.0.0 | 监听地址 |
| PORT | 8000 | 监听端口 |
| MONGODB_URL | mongodb://localhost:27017 | MongoDB连接 |
| MONGODB_DB | orchestration | 数据库名 |
| REDIS_URL | redis://localhost:6379 | Redis连接 |
| MAX_PARALLEL | 10 | 最大并行节点数 |
| LOG_LEVEL | INFO | 日志级别 |

### Docker Compose

```yaml
services:
  orchestration:
    ports:
      - "8000:8000"
    environment:
      - MONGODB_URL=mongodb://mongo:27017
      - REDIS_URL=redis://redis:6379
    depends_on:
      - mongo
      - redis
```

## 开发

### 运行测试

```bash
python -m pytest tests/ -v
```

### 项目结构

```
src/orchestration/
├── engine/          # 核心引擎
│   ├── parser.py    # 流程解析器
│   ├── scheduler.py  # DAG调度器
│   ├── context.py   # 执行上下文
│   └── engine.py    # 执行引擎
├── nodes/           # 节点类型
├── runtime/         # 运行时
├── storage/         # 数据存储
├── api/             # REST API
└── utils/           # 工具
```

## 许可证

MIT