前言

本章主要讲述：Jenkins_API接口调用

这里会简单讲下：
	1、自己使用request调用JenkinsAPI
	2、利用已有的jekinsapi库，简单封装了一个jenkins调用的类
1
2
3

• ps：之前其实简单说过【持续集成】和【持续交付】的含义，其实两者没有特别大的差别，只是所处阶段不一样 ，持续集成主要是开发阶段 ，持续交付可以理解为预发布阶段

  具体可看：https://blog.csdn.net/Makasa/article/details/125964291?spm=1001.2014.3001.5502

---

---

一、Jenkins_API简要介绍

• Jenkins_API：即Jenkins对外暴露的动作交互入口，为外部程序提供入口，可以控制Jenkins

  支持协议：Http

  常用功能：运行Job，查看任务状态，返回任务编号

---

---

二、request调用JenkinsAPI

• 这里简单介绍下request进行调用JenkinsAPI，详细看下注释，其实很简单

"""
    该类调用jenkins_api接口
        1、获取任务的最新编号
        2、获取任务的详细信息
"""
import json

import requests

# 注意：这个地址前面部分
# mikasa:yy1998123 是你的jenkins用户名和密码
# 127.0.0.1:8080 是本地jenkins域名+端口号
url = "http://mikasa:yy1998123@127.0.0.1:8080/jenkins/job/"


def get_jenkins_url(job_name):
    """
    拼接url+任务job
    :return:
    """
    # print("拼接url为：", url + job_name)
    return url + job_name


def send_api(req, tools="requests"):
    """
    对发送接口测试的工具进行封装（可以使用urlib3/requests）
    :param tools:
    :param req:
    :return:
    """
    if tools == "requests":
        return requests.request(**req)


def get_latest_job_number():
    """
    1、获取最新任务编号
    :return:
    """
    req = {
        "method": "GET",
        "url": get_jenkins_url("mikasa_demo001") + "/lastBuild/buildNumber",
    }
    res = send_api(req)
    print("最新任务编号：", res.json())


def get_job_info():
    """
    2、获取job详细信息
    :return:
    """
    req = {
        "method": "GET",
        "url": get_jenkins_url("mikasa_demo001") + "/api/json",
    }
    res = send_api(req)
    res_json = json.dumps(res.json(), indent=2)
    print("返回结果：", res_json)


get_latest_job_number()
get_job_info()
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

---

---

三、使用jenkins api库调用

• 这里介绍下利用jenkins api库去调用，上面是我们自己去写request请求去调用，而其实目前已经存在了轮子，我们直接使用Jenkins api即可

1、下载jenkinsapi库

---

2、封装jenkins调用

2.1、jenkins_api.py

"""
    封装jenkins调用类
"""
import configparser
import datetime
import logging
import os
import re
from jenkinsapi.jenkins import Jenkins

logging.basicConfig(level=logging.INFO, format='[%(asctime)s] - [%(name)s] - [%(levelname)s] - [%(message)s]')
log = logging.getLogger(__name__)


def get_jenkins_config(chose):
    """
    读取Jenkins配置：从配置文件中jenkins_server.ini
    :param chose:
    :return:
    """
    config = configparser.ConfigParser()
    # 读取配置
    config.read(os.path.join(os.getcwd(), 'jenkins_server.ini'))
    username = config.get(chose, 'username')
    password = config.get(chose, 'password')
    host = config.get(chose, 'host')
    port = config.get(chose, 'port')
    url = "http://" + host + ":" + port
    return url, username, password


class JenkinsDemo:
    def __init__(self, job_name, chose='jenkins'):
        """
        初始化，拿到jenkins配置
        :param job_name:
        :param chose:
        """
        self.job_name = job_name
        config = get_jenkins_config(chose)
        print("config:", config)
        # 解包元祖
        self.jk = Jenkins(*config, useCrumb=True)

    def __get_job_from_keys(self):
        """
        拿到所有的job名称
        :return: 返回一个列表
        """
        choose_list = []
        print(self.jk.keys())
        for my_job_name in self.jk.keys():
            # 遍历拿到所有的job，判断当前job是否在job列表里面，在的话添加到自定义列表
            if self.job_name in my_job_name:
                choose_list.append(my_job_name)
        return choose_list

    def __job_build(self, my_job_name):
        """
        构建job
        :param my_job_name:
        :return:
        """
        if self.jk.has_job(my_job_name):
            # 如果有这个job拿到他里面的job对象
            my_job = self.jk.get_job(my_job_name)
            if not my_job.is_queued_or_running():
                # 如果job当前没有在运行的话，就运行
                try:
                    # 若当前没有在跑的话，拿到最后一次构建数
                    last_build = my_job.get_last_buildnumber()
                except:
                    # 若没有获取到最后一次构建数的话，默认置为0
                    last_build = 0
                # 最新构建数+1
                build_num = last_build + 1
                try:
                    # 开始打包
                    self.jk.build_job(my_job_name)
                except Exception as e:
                    log.error(str(e))

                # 循环判断Jenkins是否打包完成
                while True:
                    # 若当前任务没有运行才获取信息
                    if not my_job.is_queued_or_running():
                        # 拿到最新一次的大奥信息
                        count_build = my_job.get_build(build_num)
                        # 获取打包开始时间
                        start_time = count_build.get_timestamp() + datetime.timedelta(hours=8)
                        # 获取打包日志
                        console_out = count_build.get_console()
                        # 获取打包状态
                        status = count_build.get_status()
                        # 获取变更内容
                        change = count_build.get_changeset_items()
                        log.info(" " + str(start_time) + " 发起的" + my_job_name + "构建已经完成，构建的状态为：" + status)
                        p2 = re.compile(r".*ERROR.*")
                        err_list = p2.findall(console_out)
                        log.info("打包日志为：" + str(console_out))
                        if status == "SUCCESS":
                            if len(change) > 0:
                                for data in change:
                                    for file_list in data["affectedPaths"]:
                                        log.info("发起的" + my_job_name + "变更的类：" + file_list)
                                    log.info("发起的" + my_job_name + "变更的备注：" + data["msg"])
                                    log.info("发起的" + my_job_name + "变更的提交人：" + data["author"]["fullName"])
                            else:
                                log.info("发起的" + my_job_name + "构建没有变更内容")

                            if len(err_list) > 0:
                                log.warning("构建的" + my_job_name + "构建状态为成功，但包含了以下错误：")
                                for error in err_list:
                                    log.error(error)
                        else:
                            if len(err_list) > 0:
                                log.warning("构建的" + my_job_name + "包含了以下错误：")
                                for error in err_list:
                                    log.error(error)
                        break
            else:
                log.warning("发起的" + my_job_name + "Jenkins is running")
        else:
            log.warning("发起的" + my_job_name + "没有该服务")

    def run(self):
        my_job_name = self.__get_job_from_keys()
        if len(my_job_name) == 1:
            self.__job_build(my_job_name[0])
        elif len(my_job_name) == 0:
            log.error("输入的job名称不正确！")


if __name__ == '__main__':
    jk = JenkinsDemo("mikasa_demo001")
    jk.run()
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

---

2.2、jenkins_server.ini

[jenkins]
username=mikasa
password=yy1998123
host=127.0.0.1
port=8080
1
2
3
4
5