feat/core python api

.gitignore

@@ -56,3 +56,5 @@ FEATURE_SUMMARY.md
 GEMINI.md
 QWEN.md
 .omx/
+.opencode/
+CODEBUDDY.md

swanlab/sdk/internal/core_python/api/__init__.py

@@ -3,6 +3,28 @@
 @file: __init__.py
 @time: 2026/3/7 18:19
 @description: SwanLab 运行时 API 封装
+
 绝大多数API使用 Client 对象，少部分API使用requests库直接调用
 我们以rpc风格封装API，方便调用
 """
+
+from .experiment import (
+    create_or_resume_experiment,
+    get_project_experiments,
+    get_single_experiment,
+    send_experiment_heartbeat,
+    update_experiment_state,
+)
+from .project import get_or_create_project, get_project
+
+__all__ = [
+    # experiment
+    "create_or_resume_experiment",
+    "get_project_experiments",
+    "get_single_experiment",
+    "send_experiment_heartbeat",
+    "update_experiment_state",
+    # project
+    "get_project",
+    "get_or_create_project",
+]

swanlab/sdk/internal/core_python/api/experiment.py

@@ -5,16 +5,17 @@
 @description: SwanLab 运行时实验API
 """
 
-from typing import List, Literal, Optional
+from typing import Dict, List, Literal, Optional, Union
 
 from google.protobuf.timestamp_pb2 import Timestamp
 
 from swanlab.exceptions import ApiError
 from swanlab.proto.swanlab.run.v1.run_pb2 import RUN_STATE_ABORTED, RUN_STATE_CRASHED, RunState
 from swanlab.sdk.internal.core_python import client
 from swanlab.sdk.internal.pkg import helper
-from swanlab.sdk.typings.core_python.api.experiment import InitExperimentType
-from swanlab.sdk.typings.run import ResumeType
+from swanlab.sdk.typings.core_python.api.experiment import ExperimentType, InitExperimentType
+from swanlab.sdk.typings.run import ResumeType, RunStateType
+from swanlab.utils.experiment import parse_filter
 
 
 def create_or_resume_experiment(
@@ -83,11 +84,75 @@ def stop_experiment(username: str, project: str, cuid: str, *, state: RunState,
         this_state = "CRASHED"
     elif state == RUN_STATE_ABORTED:
         this_state = "ABORTED"
-    client.put(
+    resp = client.put(
         f"/project/{username}/{project}/runs/{cuid}/state",
         {
             "state": this_state,
             "finishedAt": finished_at.ToDatetime().isoformat() + "Z",
             "from": "sdk",
         },
     )
+    return resp.raw.status_code == 201
+
+
+def send_experiment_heartbeat(*, cuid: str, flag_id: str) -> None:
+    """
+    发送实验心跳，保持实验处于活跃状态
+    :param cuid: 实验唯一标识符
+    :param flag_id: 实验标记ID
+    """
+    client.post(f"/house/experiments/{cuid}/heartbeat", {"flagId": flag_id})
+
+
+def update_experiment_state(
+    *,
+    username: str,
+    projname: str,
+    cuid: str,
+    state: RunStateType,
+    finished_at: Optional[str] = None,
+) -> None:
+    """
+    更新实验状态
+    :param username: 实验所属用户名
+    :param projname: 实验所属项目名称
+    :param cuid: 实验唯一标识符
+    :param state: 实验状态
+    :param finished_at: 实验结束时间，格式为 ISO 8601，如果不提供则使用当前时间
+    """
+    put_data = {
+        "state": state,
+        "finishedAt": finished_at,
+        "from": "sdk",
+    }
+    put_data = {k: v for k, v in put_data.items() if v is not None}
+    client.put(f"/project/{username}/{projname}/runs/{cuid}/state", put_data)
+
+
+def get_project_experiments(
+    *,
+    path: str,
+    filters: Optional[Dict[str, object]] = None,
+) -> Union[List[ExperimentType], Dict[str, List[ExperimentType]]]:
+    """
+    获取指定项目下的所有实验信息
+    若有实验分组，则返回一个字典，使用时需递归展平实验数据
+    :param path: 项目路径 username/project
+    :param filters: 筛选实验的条件，可选。支持以下特殊 key：
+        - 'group': 按分组名称筛选，值为字符串
+        - 'tags': 按标签筛选，值为字符串列表
+        - 'name': 按实验名筛选，值为字符串
+        - 'username': 按创建人筛选，值为字符串
+        - 'job_type': 按任务类型筛选，值为字符串
+    """
+    parsed_filters = [parse_filter(k, v) for k, v in (filters or {}).items()]
+    return client.post(f"/project/{path}/runs/shows", data={"filters": parsed_filters}).data
+
+
+def get_single_experiment(*, path: str) -> ExperimentType:
+    """
+    获取指定实验信息
+    :param path: 实验路径 username/project/expid
+    """
+    proj_path, expid = path.rsplit("/", 1)
+    return client.get(f"/project/{proj_path}/runs/{expid}").data

swanlab/sdk/internal/core_python/api/project.py

@@ -16,7 +16,7 @@
 
 def get_project(*, username: str, name: str) -> ProjectType:
     """
-    获取项目信息
+    获取项目详情信息
     :param username: 项目所属的用户名
     :param name: 项目名称
     :return: 项目信息

swanlab/sdk/typings/core_python/api/__init__.py

@@ -5,3 +5,21 @@
 @description: SwanLab API类型提示
 所有后端响应类型命名以 Response 结尾
 """
+
+from .common import LabelType
+from .experiment import RunType
+from .project import InitProjectType, ProjectType, ProjResponseType
+from .user import ApiKeyType, GroupType, SelfHostedInfoType
+from .workspace import WorkspaceInfoType
+
+__all__ = [
+    "RunType",
+    "ProjectType",
+    "InitProjectType",
+    "ProjResponseType",
+    "LabelType",
+    "GroupType",
+    "ApiKeyType",
+    "SelfHostedInfoType",
+    "WorkspaceInfoType",
+]

swanlab/sdk/typings/core_python/api/common.py

@@ -0,0 +1,7 @@
+"""Common Type Definition"""
+
+from typing import TypedDict
+
+
+class LabelType(TypedDict):
+    name: str

swanlab/sdk/typings/core_python/api/experiment.py

@@ -1,13 +1,38 @@
 """
 @author: cunyue
 @file: experiment.py
-@time: 2026/4/18 17:43
+@time: 2026/3/10 19:02
 @description: SwanLab 运行时实验API类型
 """
 
-from typing import TypedDict
+from typing import List, Tuple, TypedDict
+
+from swanlab.sdk.typings.run import RunStateType
+
+from .common import LabelType
 
 
 class InitExperimentType(TypedDict):
     # 实验cuid
     cuid: str
+
+
+class ExperimentType(TypedDict):
+    # 实验名称
+    name: str
+    # 实验CUID, 唯一标识符
+    cuid: str
+    # 实验描述
+    description: str
+    # 实验标签列表
+    labels: List[LabelType]
+    # 实验状态
+    state: RunStateType
+    # 实验组
+    cluster: str
+    # 任务类型
+    job: str
+    # 创建时间
+    created_at: str
+    # 实验颜色
+    colors: Tuple[str, str]

swanlab/sdk/typings/core_python/api/project.py

@@ -5,7 +5,9 @@
 @description: SwanLab 运行时项目API类型
 """
 
-from typing import Literal, TypedDict
+from typing import Dict, List, TypedDict
+
+from swanlab.sdk.typings.run import VisibilityType
 
 
 class _ProjectCount(TypedDict):
@@ -27,7 +29,7 @@ class ProjectType(TypedDict):
     # 项目路径 '/:username/:name'
     path: str
     # 项目可见性
-    visibility: Literal["PUBLIC", "PRIVATE"]
+    visibility: VisibilityType
     # 项目统计信息
     _count: _ProjectCount
 

swanlab/sdk/typings/run/__init__.py

@@ -27,3 +27,39 @@
 """
 异步日志上报类型
 """
+
+WorkspaceType = Literal["TEAM", "PERSON"]
+"""
+工作空间类型
+"""
+
+VisibilityType = Literal["PUBLIC", "PRIVATE"]
+"""
+项目可见类型
+"""
+
+RoleType = Literal["VISITOR", "VIEWER", "MEMBER", "OWNER"]
+"""
+组织成员类型
+"""
+
+IdentityType = Literal["root", "user"]
+"""
+Self-Hosted 用户身份类型
+"""
+
+LicensePlanType = Literal["free", "commercial"]
+"""
+Self-Hosted 许可证类型
+"""
+
+
+RunStateType = Literal["RUNNING", "FINISHED", "CRASHED", "ABORTED", "OFFLINE"]
+"""
+实验状态类型
+"""
+
+ColumnType = Literal["SCALAR", "CONFIG", "STABLE"]
+"""
+列类型
+"""

swanlab/utils/__init__.py

@@ -6,6 +6,23 @@
 理论上本模块的内容都可以被用户调用，并被写入API文档中
 """
 
-from .experiment import generate_color, generate_id, generate_name
+from .column import parse_column_type, to_camel_case
+from .experiment import (
+    extract_part_urls,
+    extract_upload_id,
+    generate_color,
+    generate_id,
+    generate_name,
+    unwrap_api_payload,
+)
 
-__all__ = ["generate_color", "generate_id", "generate_name"]
+__all__ = [
+    "generate_color",
+    "generate_id",
+    "generate_name",
+    "to_camel_case",
+    "parse_column_type",
+    "unwrap_api_payload",
+    "extract_upload_id",
+    "extract_part_urls",
+]

swanlab/utils/column/__init__.py

@@ -0,0 +1,17 @@
+from swanlab.sdk.typings.run import ColumnType
+
+
+def parse_column_type(column: str) -> ColumnType:
+    """从前缀中获取指标类型"""
+    column_type = column.split(".", 1)[0]
+    if column_type == "summary":
+        return "SCALAR"
+    elif column_type == "config":
+        return "CONFIG"
+    else:
+        return "STABLE"
+
+
+def to_camel_case(name: str) -> str:
+    """将下划线命名转化为驼峰命名"""
+    return "".join([w.capitalize() if i > 0 else w for i, w in enumerate(name.split("_"))])