阿里云效流水线API简单使用

# -*- 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()