一、Jenkins API 接口

一、Jenkins API简介

• Jenkins对外暴露的动作交互入口
• 为外部程序提供入口，可以控制Jenkins，
• 支持协议- Http
• API接口支持用户名、密码认证
• Jenkins API支持的典型功能：运行Job，查看任务状态，返回任务编号，......，等

二、Jenkins API环境准备

• 创建一个有任务运行和状态查询权限的用户
• 准备一个打算通过API远程控制的任务
• 较老版本的 jenkins 关闭跨站脚本伪造请求保护，新的采取Crumb

image.png

三、远程调用Jenkins API启动任务

• 任务名：first_ job
• 远程API服务地址：http://host:8080/job/first_job/build
• 请求方法：POST
• 用户名、密码添加方法：username:password@hostname:port ...
• 运行期望结果：
  • 任务启动
  • 服务返回 http status：201
• 使用python 的 requests 库或者使用postman进行调用

远程调用Jenkins API返回最新任务编号

• 任务名：first_job
• 远程API服务地址：http://host:8080/job/first_job/lastBuild/buildNumber
• 请求方法：GET
• 用户名、密码添加方法：username:password@hostname:port ....
• 运行期望结果：
  • 任务启动
  • 服务返回http status：201
• python脚本：

import requests
url = "http://username:password@host:8080/job/first_job/lastBuild/buildNumber"
res = requests.get(url)
print(res)

远程调用Jenkins API查询任务状态

• 任务名：first_job
• 远程API服务地址：http://host:8080/job/first_job/<build number>/api/json
• 请求方法：GET
• 用户名、密码添加方法：username: password@hostname:port ...
• 运行期望结果：
  • 任务详情JSON
  • 服务返回http status：200
• python脚本：

import requests
import json
url = "http://username:password@host:8080/job/first_job/<build number>/api/json"
res = requests.get(url)
print(json.dumps(res.json(), indent=2))

image.png

这里的<build number>就是上一个api返回的任务编号

四、jenkinsapi库

• 前面所讲的几个链接可以自己去封装api调用

• 由于已经存在了轮子，故也能直接使用已经造好的轮子

• pip install jenkinsapi

• from jenkinsapi.jenkins import Jenkins

• jk =Jenkins(url, username, password, useCrumb=True)

• 这样就能进行api的调用了

• python 文件

#!/user/bin/env python
# -*- coding: utf-8 -*-
import configparser
import datetime
import logging
import os
import re
from jenkinsapi.jenkins import Jenkins


logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(filename)s %(levelname)s %(message)s',
                    datefmt='%a %d %b %Y %H:%M:%S',
                    filename='my.log',
                    filemode='w')

log = logging.getLogger(__name__)

def get_jk_config(chose):
    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'):
        self.job_name = job_name
        config = get_jk_config(chose)
        self.jk = Jenkins(*config, useCrumb=True)

    def __get_job_from_keys(self):
        choose_list = []
        print(self.jk.keys())
        for my_job_name in self.jk.keys():
            if self.job_name in my_job_name:
                choose_list.append(my_job_name )
        return choose_list

    def __job_build(self, my_job_name):
        if self.jk.has_job(my_job_name):
            my_job = self.jk.get_job(my_job_name)
            if not my_job.is_queued_orrunning() :
                try:
                    last_build = my_job.get_last_buildnumber()
                except:
                    last_build = 0
                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 + "构建已经完成，构建的")
                        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("first_job")
    jk.run()

• jenkins_server.ini

[jenkins]
username=test_b
password=aa123456
host=106.53.223.46
port=8080