数据开发任务流特殊接口

涵盖 ETL 任务流（/etl_pipeline_modelview/api）、ETL 运行实例（/etl_pipeline_run_modelview/api）、服务化任务流（/service_pipeline_modelview/api）。

ETL任务流 /etl_pipeline_modelview

代码文件：myapp/views/view_etl_pipeline.py（route_base = '/etl_pipeline_modelview/api'）

获取/保存 Pipeline 配置信息

GET   /etl_pipeline_modelview/api/config/<etl_pipeline_id>    # 获取
POST  /etl_pipeline_modelview/api/config/<etl_pipeline_id>    # 保存

响应：

{
  "status": 0,
  "id": 1,
  "name": "etl-demo",
  "label": "ETL示例",
  "project": "demo-project",
  "pipeline_ui_config": {...},
  "pipeline_jump_button": [...],
  "pipeline_run_button": [...],
  "dag_json": {...},
  "config": {...},
  "type_warnings": [...]
}

说明：

• POST 时合并前端和后端的 dag_json（保留后端字段）
• 自动校验上游节点合法性（移除不存在的上游引用）
• 进行 I/O 类型校验（validate_dag_types()）

ETL 模板列表

GET  /etl_pipeline_modelview/api/template/list/
GET  /etl_pipeline_modelview/api/template/list/<etl_pipeline_id>

响应： JSON，包含模板分组和每个模板的 UI 配置参数。

提交 ETL 任务

GET/POST  /etl_pipeline_modelview/api/submit_etl_pipeline/<etl_pipeline_id>

响应： 302 重定向到 ETL 任务列表页面

说明： 调用 workflow handler 的 submit_pipeline() 方法调度/执行 Pipeline。

ETL 编排界面

GET  /etl_pipeline_modelview/api/web/<etl_pipeline_id>

响应： 302 重定向到 VisonPlus 可视化编排器

批量下载

GET  /etl_pipeline_modelview/api/download/
GET  /etl_pipeline_modelview/api/download/<int:etl_pipeline_id>

响应： JSON 文件下载（etl_pipeline-{timestamp}.json）

说明：

• 无参数版本：按 filter 批量导出（上限 20000）
• 带 ID 版本：导出指定 Pipeline（仅管理员/创建者/项目成员可导出）
• 导出格式：{"pipeline_name": {"pipeline": {...}}}

批量上传

POST  /etl_pipeline_modelview/api/upload/
Content-Type: multipart/form-data

响应：

{"status": 0, "message": "result为上传成功行，共成功2条", "result": ["success", "错误信息"]}

说明： 同名 Pipeline 会覆盖。导入时自动匹配项目名（不存在则报错）。

下载上传模板

GET  /etl_pipeline_modelview/api/download_template/

响应： JSON 文件下载（etl_pipeline_template.json）

ETL运行实例 /etl_pipeline_run_modelview

代码文件：myapp/views/view_etl_pipeline_run.py（route_base = '/etl_pipeline_run_modelview/api'）

Run 平台告警列表

GET  /etl_pipeline_run_modelview/api/<run_id>/alerts

响应：

{"status": 0, "result": [{...alert_object...}], "count": 5}

说明： 读取 {log_path}/alerts.jsonl 文件（每行一个 JSON 告警对象）。

Run 全节点预览汇总

GET  /etl_pipeline_run_modelview/api/<run_id>/preview

响应：

{"status": 0, "result": {"task_id_1": {...preview...}, "task_id_2": {...preview...}}}

说明： 扫描所有子目录，读取各节点的 preview.json。

单节点预览

GET  /etl_pipeline_run_modelview/api/<run_id>/preview/<task_id>

响应：

{"status": 0, "result": {...preview_json...}}

最近一次运行的某节点预览

GET  /etl_pipeline_run_modelview/api/preview_latest/<int:pipeline_id>/<task_id>

响应：

{"status": 0, "result": {"run_id": "xxx", "preview": {...}}}

单节点日志

GET  /etl_pipeline_run_modelview/api/<run_id>/node_log/<task_id>

响应： HTML（暗色主题格式化日志）。读取 {log_path}/{task_id}/node.log，超过 1MB 截取最后 1MB。

⚠️ 已按代码改正：原文档此处写的是 “Run 主日志 GET /<run_id>/log”，但当前代码中不存在 /<run_id>/log 路由；日志按节点粒度提供，路由为 /<run_id>/node_log/<task_id>，见 myapp/views/view_etl_pipeline_run.py:274（node_log）。

单节点数据文件下载

GET  /etl_pipeline_run_modelview/api/<run_id>/data_file/<task_id>/<filename>

响应： 文件下载（.csv → text/csv，其余 → application/json）。filename 仅允许 [a-zA-Z0-9_\-\.]+。见 myapp/views/view_etl_pipeline_run.py:292。

停止 Run

POST  /etl_pipeline_run_modelview/api/<run_id>/stop

响应：

{"status": 0, "message": "已停止"}

说明： 删除 K8s Job/Pod/ConfigMap，设置 run.status = 'stopped'。仅管理员/创建者/项目成员可操作。

Run 当前状态

GET  /etl_pipeline_run_modelview/api/<run_id>/status

响应：

{"status": 0, "result": {"status": "running", "duration": "2m30s", "host": "node-1", "pod_name": "pod-xxx"}}

DAG 执行跟踪

GET  /etl_pipeline_run_modelview/api/<run_id>/web/dag

响应：

{
  "status": 0,
  "result": {
    "control": {"node_ops": [...], "direction": "vertical"},
    "dag": [{...node_tree...}],
    "layout": {...overall_metadata...}
  }
}

说明： 从 progress.json 读取实时状态，构建 DAG 树结构，映射任务状态（pending/running/success/failed/skipped）。与 /workflow_modelview/api/web/dag 格式对齐，可复用前端 /frontend/commonRelation 展示跟踪图。

节点详情

GET  /etl_pipeline_run_modelview/api/<run_id>/web/node_detail/<task_id>

响应：

{
  "status": 0,
  "result": {
    "detail": [
      {"tabName": "数据预览", "content": [...], "bottomButton": [...]},
      {"tabName": "日志", "content": [...]},
      {"tabName": "节点信息", "content": [...]}
    ]
  }
}

说明： 支持多种预览类型：dataframe（HTML 表格）、model、dict、scalar、skipped。

服务化任务流 /service_pipeline_modelview

代码文件：myapp/views/view_service_pipeline.py（route_base = '/service_pipeline_modelview/api'）

我的服务化任务流列表

GET  /service_pipeline_modelview/api/my/list/

无需参数。返回当前用户的所有服务化 Pipeline。

响应：

{"status": 0, "message": "success", "result": [{...service_pipeline_json...}]}

运行服务化任务流

GET/POST  /service_pipeline_modelview/api/run_service_pipeline/<service_pipeline_id>

响应： 302 重定向到日志页面

说明：

• 仅管理员或项目成员可执行
• 先清理之前的部署
• 根据根节点模板类型选择部署方式：mq（Kafka 消费者）或 gateway（HTTP 端点）
• 创建 K8s Deployment 和 ConfigMap

服务化任务流编排界面

GET  /service_pipeline_modelview/api/web/<service_pipeline_id>

响应： 302 重定向到可视化编排器

服务化任务流日志界面

GET  /service_pipeline_modelview/api/web/log/<service_pipeline_id>

响应： HTML（暂未实现）

服务化任务流监控界面

GET  /service_pipeline_modelview/api/web/monitoring/<service_pipeline_id>

响应： 302 重定向到 Grafana 仪表盘

服务化任务流 Pod 界面

GET  /service_pipeline_modelview/api/web/pod/<service_pipeline_id>

响应： 302 重定向到 K8s Pod 搜索页面

清理服务化任务流

GET/POST  /service_pipeline_modelview/api/clear/<service_id>

响应： 302 重定向到列表页面

说明： 删除 K8s Deployment。

获取/保存服务化任务流配置

GET   /service_pipeline_modelview/api/config/<service_pipeline_id>    # 获取
POST  /service_pipeline_modelview/api/config/<service_pipeline_id>    # 保存

请求/响应格式同 ETL Pipeline 的 config 接口。

服务化任务流模板列表

GET  /service_pipeline_modelview/api/template/list/

响应：

{
  "status": 0,
  "message": "success",
  "templte_common_ui_config": {...},
  "template_group_order": ["入口", "逻辑节点", "功能节点"],
  "templte_list": {
    "入口": [{"template_name": "kafka", ...}],
    "逻辑节点": [{"template_name": "switch", ...}],
    "功能节点": [{"template_name": "http", ...}, {"template_name": "custom_method", ...}]
  }
}