使用阿里云效API操作流水线

使用阿里云效（Alibaba Cloud DevOps）API操作流水线时，需要注意以下几个方面：

1. 认证与授权
   确保你已经获取了正确的访问凭证（AccessKey ID 和 AccessKey Secret），并且这些凭证具有足够的权限来执行你需要的操作。可以通过阿里云的RAM（资源访问管理）控制台管理和分配这些权限。

2. API 调用限制
   阿里云效API通常有调用频率限制（QPS）。确保在设计你的应用程序时考虑到这些限制，以避免超出配额导致的调用失败。

3. 错误处理
   在使用API时，要做好错误处理和异常捕获。阿里云效API会返回详细的错误信息，包括错误代码和错误消息。需要根据这些信息来诊断问题并采取适当的措施。

4. API版本
   确保你使用的API版本是最新的，并且在代码中指定了正确的版本。如果API有更新或废弃，需及时更新你的代码。

5. 参数验证
   在调用API前，确保所有必需参数都已正确设置，并且这些参数的值符合API文档中的要求。

6. 安全性
   在代码中不要硬编码敏感信息如AccessKey ID和AccessKey Secret。可以使用环境变量或其他安全存储方式来管理这些信息。

7. 调试和日志
   启用详细的日志记录和调试信息，以便在出现问题时能够快速定位和解决问题。

8. 资源管理
   确保在使用API创建或修改资源时，有相应的清理逻辑，以避免不必要的资源浪费。

注意事项：

1. 阿里云子账户需要给策略权限： AliyunRDCFullAccess
2. 子账号需要加入云效企业内部，并给足操作权限

文档： 云效API接口 ，云效错误码

示例代码

1. 使用python 创建 Pipeline_sample.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import json
import os
import sys

from typing import List

from alibabacloud_devops20210625.client import Client as devops20210625Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
from alibabacloud_devops20210625 import models as devops_20210625_models
from alibabacloud_tea_console.client import Client as ConsoleClient

ORGANIZATION_ID = [云效企业ID]


class Pipeline:

    def __init__(self):
        self.organization_id = ORGANIZATION_ID
        pass

    @staticmethod
    def create_client() -> devops20210625Client:
        """
        使用AK&SK初始化账号Client
        @return: Client
        @throws Exception
        """
        # 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
        # 建议使用更安全的 STS 方式，更多鉴权访问方式请参见：https://help.aliyun.com/document_detail/378659.html。
        config = open_api_models.Config(
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # 必填，请确保代码运行环境设置了环境变量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # Endpoint 请参考 https://api.aliyun.com/product/devops
        config.endpoint = f'devops.cn-hangzhou.aliyuncs.com'
        return devops20210625Client(config)

    """
    create_pipeline
    """

    @staticmethod
    def create_pipeline(name: str, yaml_file_path: str) -> None:
        # 读取 YAML 文件内容
        with open(yaml_file_path, 'r') as file:
            content = file.read()

        client = Pipeline.create_client()
        create_pipeline_request = devops_20210625_models.CreatePipelineRequest(
            name=name,
            content=content
        )
        runtime = util_models.RuntimeOptions()
        headers = {}
        try:
            # 复制代码运行请自行打印 API 的返回值
            response = client.create_pipeline_with_options(ORGANIZATION_ID, create_pipeline_request, headers,
                                                           runtime)
            # 如果API调用成功，这里将处理并打印返回的数据
            if response:
                print("Pipeline created successfully.")
                # 假设返回的数据结构中包含一个status或类似字段来表示操作状态
                print("Response Data:", response)
                # print(response.body.__dict__)
                # 提取并打印 pipelinId
                pipelinId = response.body.pipelin_id
                print(pipelinId)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    def delete_pipeline(pipelin_id: str) -> None:
        client = Pipeline.create_client()
        runtime = util_models.RuntimeOptions()
        headers = {}
        try:
            # 复制代码运行请自行打印 API 的返回值
            response = client.delete_pipeline_with_options(ORGANIZATION_ID, pipelin_id, headers, runtime)
            # 如果API调用成功，这里将处理并打印返回的数据
            if response:
                print("Pipeline created successfully.")
                # 假设返回的数据结构中包含一个status或类似字段来表示操作状态
                print("Response Data:", response)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    def get_pipelines_list() -> None:
        client = Pipeline.create_client()
        list_pipelines_request = devops_20210625_models.ListPipelinesRequest(
            max_results=10,
            next_token='123'
        )
        runtime = util_models.RuntimeOptions()
        headers = {}
        try:
            # 复制代码运行请自行打印 API 的返回值
            response = client.list_pipelines_with_options(ORGANIZATION_ID, list_pipelines_request, headers, runtime)
            # 如果API调用成功，这里将处理并打印返回的数据
            if response:
                print("Pipeline created successfully.")
                # 假设返回的数据结构中包含一个status或类似字段来表示操作状态
                print("Response Data:", response)
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    def run_pipeline(json_file_path: str, pipeline_id: str) -> None:
        with open(json_file_path, 'r') as file:
            params = json.load(file)

        client = Pipeline.create_client()
        start_pipeline_run_request = devops_20210625_models.StartPipelineRunRequest(
            # params='''{"envs": {
            #                         "UPDATE": "ALL",
            #                         "IMAGES":"sjdlkfjslkdajf"
            #                     }}'''
            params=params
        )
        runtime = util_models.RuntimeOptions()
        headers = {}
        try:
            resp = client.start_pipeline_run_with_options(ORGANIZATION_ID, pipeline_id,start_pipeline_run_request, headers, runtime)
            ConsoleClient.log(UtilClient.to_jsonstring(resp))
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':

    # 创建一个测试流水线
    # Pipeline.create_pipeline( name='测试流水线test4', yaml_file_path='pipeline_content.yaml')
    # 删除指定的流水线
    # Pipeline.delete_pipeline('3263926')
    # 执行指定的流水线
    Pipeline.run_pipeline("pipeline_params.json","3260926")
    # 获取流水线列表
    # Pipeline.get_pipelines_list()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

传参文件 pipeline_params.json

{
    "envs": {
        "UPDATE": "ALL",
        "IMAGES": "sjdlkfjslkdajf"
    }
}
1
2
3
4
5
6

流水线yaml配置 pipeline_content.yaml

文件可以再流水线使用yaml模式编辑后替换

stages:
  vm_deploy_stage:
    name: "部署"
    jobs:
      vm_deploy_job:
        name: "主机部署`测试主机"
        component: "VMDeploy"
        with:
          downloadArtifact: false
          useEncode: false
          machineGroup: "eSa9wUssQfX799kg"
          run: "/home/test.sh"
          artifactDownloadPath: ""
          executeUser: "root"
          artifact: ""
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15