DataOps-platform/test_update_dataflow.py

#!/usr/bin/env python3
"""测试更新数据流接口"""
import requests
import json

data = {
    "name_zh": "测试用销售数据",
    "name_en": "try_sales_data",
    "category": "应用类",
    "leader": "system",
    "organization": "citu",
    "script_type": "sql",
    "update_mode": "append",
    "frequency": "月",
    "tag": [],
    "status": "active",
    "script_requirement": {
        "rule": None,
        "source_table": [],
        "target_table": [241]
    }
}

response = requests.put(
    "http://localhost:5500/api/dataflow/update-dataflow/258",
    json=data,
    headers={"Content-Type": "application/json"}
)
print("Status Code:", response.status_code)
print("Response:", response.text)