Python 客户端调用示例

以 pipeline 和 task 的增删改查为例，演示标准接口的 Python 调用方式。

• Authorization 头使用 $username|$token 形式（示例中简化为 admin），token 在个人信息页面获取。
• 外键字段用关系名传参（如 pipeline、project），不是 pipeline_id，见 01-通用约定与路径对照.md。
• 末尾的 run_pipeline 为特殊接口（非标准 CRUD），见 接口总览。

import requests
import uuid
import json

host = 'http://xx.xx.xx.xx'
pipeline_base_route = 'pipeline_modelview'
task_base_route = 'task_modelview'
headers = {
  'Content-Type': 'application/json',
  'Authorization': 'admin'    # 替换为实际的 $username|$token
}

# ===== 通用 CRUD 函数 =====

def add(base_route, payload):
    url = "%s/%s/api/" % (host, base_route)
    response = requests.post(url, headers=headers, json=payload)
    if response.status_code == 200:
        back_data = response.json()
        if back_data.get('status') == 0 and 'result' in back_data:
            return back_data['result']
    return None

def edit(base_route, id, payload):
    url = "%s/%s/api/%s" % (host, base_route, str(id))
    response = requests.put(url, headers=headers, json=payload)
    if response.status_code == 200:
        back_data = response.json()
        if back_data.get('status') == 0 and 'result' in back_data:
            return back_data['result']
    return None

def get(base_route, id):
    url = "%s/%s/api/%s" % (host, base_route, str(id))
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        back_data = response.json()
        if back_data.get('status') == 0 and 'result' in back_data:
            return back_data['result']
    return None

def list_records(base_route, payload):
    url = "%s/%s/api/" % (host, base_route)
    response = requests.get(url, headers=headers, json=payload)
    if response.status_code == 200:
        back_data = response.json()
        if back_data.get('status') == 0 and 'result' in back_data:
            return back_data['result']
    return None

def delete(base_route, id):
    url = "%s/%s/api/%s" % (host, base_route, str(id))
    response = requests.delete(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    return None

# ===== 创建 pipeline =====
pipeline = add(pipeline_base_route, {
    'project': '1',
    'name': 'pipeline-' + uuid.uuid4().hex[:4],
    'describe': '测试pipeline',
    'namespace': 'pipeline',
    'schedule_type': 'once',
    'global_args': '',
    'cron_time': '1 1 * * *',
    'node_selector': "cpu=true",
    'image_pull_policy': 'Always',
    'parallelism': 3
})

# ===== 创建 task（外键绑定 pipeline） =====
task1 = add(task_base_route, {
    'pipeline': pipeline['id'],
    'job_template': 1,
    'name': 'task-' + uuid.uuid4().hex[:4],
    'label': '测试task',
    'working_dir': '',
    'command': 'sleep 10',
    'overwrite_entrypoint': 0,
    'volume_mount': "",
    'node_selector': 'cpu=true',
    'resource_memory': '300M',
    'resource_cpu': '0.3',
    'resource_gpu': '0',
    'timeout': 30000,
    'retry': 2,
    'outputs': '',
    'args': json.dumps({"aa": "aa1"})
})

task2 = add(task_base_route, {
    'pipeline': pipeline['id'],
    'job_template': 1,
    'name': 'task-' + uuid.uuid4().hex[:4],
    'label': '测试task2',
    'command': 'sleep 5',
    'args': json.dumps({"aa": "aa2"})
})

# ===== 更新 task 资源 =====
task1['resource_memory'] = '1G'
task1 = edit(task_base_route, task1['id'], task1)

# ===== 更新 pipeline DAG（设置任务依赖） =====
pipeline['dag_json'] = json.dumps({
    task1['name']: {
        "upstream": [task2['name']]
    }
})
pipeline = edit(pipeline_base_route, pipeline['id'], pipeline)

# ===== 查看 pipeline =====
pipeline = get(pipeline_base_route, pipeline['id'])

# ===== list 查询（带过滤和分页） =====
tasks = list_records(task_base_route, {
    "page": 0,
    "page_size": 10,
    "filters": [
        {"col": "pipeline", "opr": "rel_o_m", "value": pipeline['id']}
    ]
})

# ===== 运行 pipeline（特殊接口，非标准CRUD） =====
run_res = requests.get(
    '%s/%s/api/run_pipeline/%s' % (host, pipeline_base_route, pipeline['id']),
    headers=headers,
    allow_redirects=False
)
if run_res.status_code == 302:
    print('运行成功，跳转:', run_res.headers['Location'])

# ===== 删除 =====
delete(pipeline_base_route, pipeline['id'])

注意：上面 add pipeline 时传入的 resource_memory='300M' 等字段需符合各模型的校验规则（如 task 的 resource_memory 正则为 ^[0-9]*G$，示例中的 300M/1G 仅为演示，实际以 _info 返回的校验为准），各模型字段详情见 标准化接口参数详情。