DataOps-platform/docs/API_get_BD_list接口说明.md

GET /api/dataflow/get-BD-list 接口说明

接口概述

获取 Neo4j 图数据库中所有 BusinessDomain（业务域）节点的列表。

接口信息

• 路径: /api/dataflow/get-BD-list
• 方法: GET
• 认证: 需要（根据项目配置）

请求参数

无需请求参数

响应格式

成功响应

HTTP状态码: 200

响应体:

{
  "code": 200,
  "data": [
    {
      "id": 276,
      "name_zh": "科室对照表",
      "name_en": "TB_JC_KSDZB",
      "tag": "数据资源"
    },
    {
      "id": 277,
      "name_zh": "科室对照表",
      "name_en": "departments",
      "tag": "数据模型"
    }
  ],
  "message": "操作成功"
}

响应字段说明

字段	类型	说明
code	Integer	响应状态码，200表示成功
data	Array	BusinessDomain节点列表
data[].id	Integer	节点ID（Neo4j内部ID）
data[].name_zh	String	业务域中文名称
data[].name_en	String	业务域英文名称
data[].tag	String	标签名称，通过BELONGS_TO关系获取的DataLabel节点的中文名称
message	String	响应消息

错误响应

HTTP状态码: 500

响应体:

{
  "code": 500,
  "message": "获取BusinessDomain列表失败: {错误详情}",
  "data": {}
}

实现逻辑

1. 路由层（app/api/data_flow/routes.py）

@bp.route('/get-BD-list', methods=['GET'])
def get_business_domain_list():
    """获取BusinessDomain节点列表"""
    try:
        logger.info("接收到获取BusinessDomain列表请求")
        
        # 调用服务层函数获取BusinessDomain列表
        bd_list = DataFlowService.get_business_domain_list()
        
        res = success(bd_list, "操作成功")
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
    except Exception as e:
        logger.error(f"获取BusinessDomain列表失败: {str(e)}")
        res = failed(f'获取BusinessDomain列表失败: {str(e)}', 500, {})
        return json.dumps(res, ensure_ascii=False, cls=MyEncoder)

2. 服务层（app/core/data_flow/dataflows.py）

@staticmethod
def get_business_domain_list() -> List[Dict[str, Any]]:
    """
    获取BusinessDomain节点列表
    
    Returns:
        BusinessDomain节点列表，每个节点包含 id, name_zh, name_en, tag
    """
    try:
        logger.info("开始查询BusinessDomain节点列表")
        
        with connect_graph().session() as session:
            # 查询所有BusinessDomain节点及其BELONGS_TO关系指向的标签
            query = """
            MATCH (bd:BusinessDomain)
            OPTIONAL MATCH (bd)-[:BELONGS_TO]->(label:DataLabel)
            RETURN id(bd) as id, 
                   bd.name_zh as name_zh, 
                   bd.name_en as name_en,
                   label.name_zh as tag
            ORDER BY bd.create_time DESC
            """
            
            result = session.run(query)
            
            bd_list = []
            for record in result:
                bd_item = {
                    "id": record["id"],
                    "name_zh": record["name_zh"] if record["name_zh"] else "",
                    "name_en": record["name_en"] if record["name_en"] else "",
                    "tag": record["tag"] if record["tag"] else ""
                }
                bd_list.append(bd_item)
            
            logger.info(f"成功查询到 {len(bd_list)} 个BusinessDomain节点")
            return bd_list
            
    except Exception as e:
        logger.error(f"查询BusinessDomain节点列表失败: {str(e)}")
        raise e

3. Cypher 查询说明

MATCH (bd:BusinessDomain)
OPTIONAL MATCH (bd)-[:BELONGS_TO]->(label:DataLabel)
RETURN id(bd) as id, 
       bd.name_zh as name_zh, 
       bd.name_en as name_en,
       label.name_zh as tag
ORDER BY bd.create_time DESC

查询逻辑：

1. 匹配所有 BusinessDomain 节点
2. 可选匹配其通过 BELONGS_TO 关系指向的 DataLabel 节点
3. 返回 BusinessDomain 的 id、中文名、英文名
4. 返回 DataLabel 的中文名作为 tag
5. 按创建时间降序排列

使用示例

cURL 示例

curl -X GET http://localhost:5000/api/dataflow/get-BD-list \
  -H "Content-Type: application/json"

JavaScript 示例

fetch('http://localhost:5000/api/dataflow/get-BD-list', {
  method: 'GET',
  headers: {
    'Content-Type': 'application/json'
  }
})
.then(response => response.json())
.then(data => {
  console.log('BusinessDomain列表:', data.data);
  data.data.forEach(bd => {
    console.log(`ID: ${bd.id}, 名称: ${bd.name_zh} (${bd.name_en}), 标签: ${bd.tag}`);
  });
})
.catch(error => console.error('Error:', error));

Python 示例

import requests

url = "http://localhost:5000/api/dataflow/get-BD-list"
response = requests.get(url)

if response.status_code == 200:
    result = response.json()
    bd_list = result['data']
    
    for bd in bd_list:
        print(f"ID: {bd['id']}, 名称: {bd['name_zh']} ({bd['name_en']}), 标签: {bd['tag']}")
else:
    print(f"请求失败: {response.text}")

注意事项

1. 标签字段（tag）：

   • tag 字段通过查询 BusinessDomain 节点的 BELONGS_TO 关系获取
   • 如果节点没有 BELONGS_TO 关系，tag 字段为空字符串
   • 标签的值来自 DataLabel 节点的 name_zh 属性

2. 排序规则：

   • 结果按 BusinessDomain 节点的 create_time 降序排列
   • 最新创建的节点排在前面

3. 空值处理：

   • 如果 name_zh、name_en 或 tag 为 null，会转换为空字符串 ""
   • 保证前端不会收到 null 值

4. 性能考虑：

   • 该接口返回所有 BusinessDomain 节点，不支持分页
   • 如果节点数量很大，建议后续添加分页功能

相关接口

• POST /api/model/data/save - 创建数据模型时会自动创建 BusinessDomain 节点
• 其他数据流相关接口参见 /api/dataflow/ 路径下的接口

更新历史

• 2024-11-28: 初始版本，实现基本的 BusinessDomain 节点列表查询功能