API概览：https://help.aliyun.com/document_detail/460575.html?spm=a2c4g.460564.0.0.2ffe2e09elB6B5
API在线调试：https://next.api.aliyun.com/api/devops/2021-06-25/StartPipelineRun?sdkStyle=old
# -*- coding: utf-8 -*- import sys import json from aliyunsdkcore.client import AcsClient from aliyunsdkcore.request import CommonRequest class AliyunPipeline(object): def __init__(self, region="cn-hangzhou"): self.region = region # 设置阿里云的AccessKeyID和AccessKeySecret self.AK = "xxxxx" self.AS = "xxxxxx" self.client = AcsClient(self.AK, self.AS, self.region) self.request = CommonRequest() self.request.set_accept_format('json') self.request.set_protocol_type('https') # https | http self.request.set_domain('devops.cn-hangzhou.aliyuncs.com') self.request.set_version('2021-06-25') self.request.add_header('Content-Type', 'application/json') self.organization_id = "5f5ad8xxxxx" def get_pipeline_name(self, pipeline_id): self.request.set_method('GET') uri_pattern = "/organization/{}/pipelines/{}".format(self.organization_id, pipeline_id) self.request.set_uri_pattern(uri_pattern) response = self.client.do_action_with_exception(self.request) response = json.loads(response) # print(response["pipeline"]["name"]) return response["pipeline"]["name"] def pipeline_run(self, pipeline_id): self.request.set_method('POST') uri_pattern = "/organizations/{}/pipelines/{}/run".format(self.organization_id, pipeline_id) self.request.set_uri_pattern(uri_pattern) response = self.client.do_action_with_exception(self.request) response = json.loads(response) # print(response) return response.get("success") def get_pipeline_deploy_info(self, pipeline_id): pipe_name = self.get_pipeline_name(pipeline_id) deploy_state = self.pipeline_run(pipeline_id) if deploy_state: print("%s 流水线触发运行成功" % pipe_name) else: print("%s 流水线触发运行失败" % pipe_name) def main(): deploy_env = sys.argv[1] #这边是通过传参获取对应项目相关应用环境，在通过写死的流水线ID进行发布，实际也可以通过流水线分组信息自动获取相关流水线ID prod_pipeline_id_list = [] dev_pipeline_id_list = ["2369060", ] #写入流水线ID obj = AliyunPipeline() if deploy_env == "dev": pipeline_id_list = dev_pipeline_id_list elif deploy_env == "prod": pipeline_id_list = prod_pipeline_id_list else: pipeline_id_list = dev_pipeline_id_list for pipeline_id in pipeline_id_list: obj.get_pipeline_deploy_info(pipeline_id) main()
