cube-studio Python SDK 使用

平台提供了 cubestudio Python SDK，可以用代码方式对接平台的数据集、任务（任务流）、推理服务、notebook 等模块，实现自动化操作。

代码核实依据：SDK 源码位于 aihub/src/cubestudio/，本文涉及的接口路径、方法名均与该目录下 request/model_client.py、request/model.py、dataset/dataset.py、train/task.py 对照核实。

一、安装与引入

SDK 随平台镜像（如 notebook 镜像）内置。如在 notebook 中使用，需要先 source 到对应的 notebook 内核环境，才能正常使用对应内核的 SDK 与依赖：

# 在同一行命令中先 source 环境名（base, python27, python36, python37, python38, python39, python310, CubeStudio），
# 才能把包 pip 安装到指定虚拟环境。不清楚有哪些环境时，可用 conda info --envs 查看。
!source activate cube-studio && pip install pandas

引入基础包：

import json, os, time, shutil

from cubestudio.request.model_client import Client, init
from cubestudio.dataset.dataset import Dataset

二、初始化客户端

所有操作前必须先调用 init(host, username, token) 初始化全局连接信息。

HOST = 'http://xx.xx.xx.xx'      # 平台访问地址
token = '<YOUR_TOKEN>'            # 当前用户的访问 token（见下文“token 获取”）
username = 'admin'               # 与 token 对应的用户名
init(host=HOST, username=username, token=token)

初始化逻辑（request/model_client.py 的 init）：

• 用 token 作为 Authorization 请求头，向 <HOST>/users/api/ 发起查询，按 username 过滤，验证 token 与用户名是否匹配；匹配成功打印“初始化验证成功”，否则抛出 token和用户名不匹配。
• 验证通过后，会预先实例化一批内置 Client：Images / Project / Pipeline / Repository / Workflow / Task / Job_Template / MyUser / User / Notebook / InferenceService。

token 获取

token 即平台为每个用户生成的访问密钥（用户模型的 secret 字段，是用 JWT_PASSWORD 签名的 JWT，去掉了固定的 HS256 头，仅保留 payload.signature 两段，见 myapp/security.py:155-174）。请求时放入 Authorization 请求头即可（myapp/security.py:649-677 的 load_user_from_header 负责解析）。

在平台“用户管理 / 个人信息”页面可查看到当前用户的 token。用户与角色相关说明见 01-入门与权限 段。

三、数据集 SDK

可以用代码自动上传数据集到平台，用于训练和推理等。数据集对象对应服务端接口 /dataset_modelview/api（dataset/dataset.py:17）。

定义/获取一个数据集

# one() 按字段过滤取一条；不存在则用 add() 新建
dataset = Client(Dataset).one(name="coco")
if not dataset:
    dataset = Client(Dataset).add(
        name='coco',
        version='v2014',
        label='coco未标注数据集',
        describe='来自于2014年数据，未标注的coco数据集',
        icon='https://pic2.zhimg.com/80/v2-399df41d8562f8f09b98d288b97c8f8d_1440w.webp',
    )

压缩、加密、上传数据集

features = json.dumps(json.load(open('coco/coco.json')), indent=4, ensure_ascii=False)
dataset = dataset.update(path='', features=features)   # 更新数据集元信息
# 压缩：compress(zip_path, compress_dir)，把 compress_dir 目录打包为 zip
dataset.compress('coco.zip', 'vision/coco')
# 加密：encrypt(file_path, save_path, key)
dataset.encrypt('coco.zip', "coco.zip.crypt", key)
# 上传：upload(file_path_list, partition='')，可指定分区
dataset.upload('coco.zip.crypt', partition='20230201')

说明：encrypt/decrypt 使用 cryptography.fernet.Fernet(key) 做对称加解密（dataset/dataset.py:63-97），key 需为 Fernet 密钥（如 from cryptography.fernet import Fernet; key = Fernet.generate_key()），加解密两端需使用同一 key。

下载、解密、解压数据集

# 清理本地旧文件
os.remove('coco.zip.crypt') if os.path.exists('coco.zip.crypt') else ''
os.remove('coco.zip') if os.path.exists('coco.zip') else ''
shutil.rmtree('coco') if os.path.exists('coco') else ''

dataset.download(partition='20230201')                 # 按分区下载（默认下到当前工作目录）
dataset.decrypt("coco.zip.crypt", 'coco.zip', key)     # 解密
dataset.decompress('coco.zip', 'coco')                 # 解压到 coco 目录

数据集方法一览（dataset/dataset.py）：download(partition='', des_dir=None)、upload(file_path_list, partition='')、compress(zip_path, compress_dir)、decompress(zip_path, extract_dir=None)、encrypt(file_path, save_path, key)、decrypt(file_path, save_path, key)。

四、推理 SDK

引入基础包：

import time, json
from cubestudio.request.model_client import Client, init
from cubestudio.train.task import InferenceService, Project

初始化客户端（同上 init(...)），然后添加一个 inferenceservice：

inferenceservice = Client(InferenceService).add_or_update(
    service_type='serving',
    project=Client(Project).one(name='public'),
    label='sdk创建inference service',
    model_name='nginx',
    model_version='v202300801',
    images='nginx',
    resource_memory='2G',
    resource_cpu='2',
    resource_gpu='0',
    min_replicas='1',
    max_replicas='1',
    ports='80',
    volume_mount='kubeflow-user-workspace(pvc):/mnt',
    working_dir='',
    command='',
    env='',
)

部署推理服务：

print(json.dumps(inferenceservice.to_dict(), indent=4))
if inferenceservice.model_status != 'online':
    inferenceservice.deploy()

InferenceService 对应服务端 /inferenceservice_modelview/api，deploy() 实际请求 /inferenceservice_modelview/api/deploy/prod/{id}（train/task.py:136-148）。

五、训练（任务流）SDK

如在 notebook 中使用需先 source 到对应内核环境（同第一节）。引入基础包：

from cubestudio.request.model_client import Client, init
from cubestudio.train.task import Job_Template, Project, Pipeline, Task
import json

初始化客户端后：

添加一个画布（pipeline）

pipeline = Client(Pipeline).add_or_update(
    name=f'{username}-default',
    describe='sdk画布',
    project=Client(Project).one(name='public'),
)

添加一个任务（task）

job_template = Client(Job_Template).one(name="自定义镜像")
task = Client(Task).add_or_update(
    name='sdk-test1',
    label='sdk发起的任务',
    pipeline=pipeline,
    job_template=Client(Job_Template).one(name="自定义镜像"),
    timeout=3600,
    retry=0,
    args=json.dumps({
        "images": "ubuntu:20.04",
        "command": 'for i in {1..50}; do date; sleep 1; done',
        "workdir": "/",
    }),
)

运行 / 跟踪日志 / 关闭任务

task.run()              # 启动任务，请求 /task_modelview/api/run/{id}
task.log(follow=True)   # 跟踪日志（follow=True 会持续轮询）
task.stop()             # 关闭任务，请求 /task_modelview/api/clear/{id}

对照 train/task.py：Task 路径 /task_modelview/api，run() → /run/{id}、stop() → /clear/{id}、log() 读取运行 pod 日志；Pipeline 路径 /pipeline_modelview/api，run() → /run_pipeline/{id}。

六、Notebook SDK

引入基础包：

import datetime, json, random, time

from cubestudio.request.model import Model
from cubestudio.request.model_client import Client, init
from cubestudio.train.task import Notebook, Project

初始化客户端后，添加一个 notebook：

notebook = Client(Notebook).add_or_update(
    name='sdk-test',
    project=Client(Project).one(name='public'),
    describe='sdk创建notebook，并直接提供给其他平台跳转使用',
    images='ccr.ccs.tencentyun.com/cube-studio/notebook-enterprise:jupyter-ubuntu-machinelearning',
    volume_mount='kubeflow-user-workspace(pvc):/mnt',
    image_pull_policy='Always',
    resource_memory='10G',
    resource_cpu='10',
    resource_gpu='0',
    expand=json.dumps({"root": "examples/Bayesian/bayesian_network.ipynb"}),
)

启动 / 重置 notebook：

print(json.dumps(notebook.to_dict(), indent=4))
notebook.reset()        # 重置并等待重启成功，完成后打印 JupyterLab 访问地址

注意：Notebook 的服务端路径是 /notebook_modelview/sdk（区别于普通 *_modelview/api，见 train/task.py:112-133），reset() 请求 /notebook_modelview/sdk/reset/{id}，启动完成后会按 expand.root 拼出 {HOST}/notebook/jupyter/{name}/lab/tree/{root} 访问地址。

七、SDK 对象与服务端接口对照表

（依据 aihub/src/cubestudio/train/task.py、dataset/dataset.py 核实，供导航；具体以源码为准）

Client(ModelClass) 通用 CRUD 方法（request/model_client.py 的 Model_Client）：add / update / show / one / search / list / add_or_update / delete / single_action / multi_action。这些通用编程接口的更深入说明，可参见 06-二次开发 段（后端框架与 API）。