一、Jenkins API簡介

• Jenkins對外暴露的動作交互入口
• 為外部程序提供入口，可以控制Jenkins，
• 支持協(xié)議- Http
• API接口支持用戶名、密碼認證
• Jenkins API支持的典型功能：運行Job，查看任務狀態(tài)，返回任務編號，......，等

二、Jenkins API環(huán)境準備

• 創(chuàng)建一個有任務運行和狀態(tài)查詢權限的用戶
• 準備一個打算通過API遠程控制的任務
• 較老版本的 jenkins 關閉跨站腳本偽造請求保護，新的采取Crumb

image.png

三、遠程調(diào)用Jenkins API啟動任務

• 任務名：first_ job
• 遠程API服務地址：http://host:8080/job/first_job/build
• 請求方法：POST
• 用戶名、密碼添加方法：username:password@hostname:port ...
• 運行期望結果：
  • 任務啟動
  • 服務返回 http status：201
• 使用python 的 requests 庫或者使用postman進行調(diào)用

遠程調(diào)用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)

遠程調(diào)用Jenkins API查詢?nèi)蝿諣顟B(tài)

• 任務名：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調(diào)用

• 由于已經(jīng)存在了輪子，故也能直接使用已經(jīng)造好的輪子

• pip install jenkinsapi

• from jenkinsapi.jenkins import Jenkins

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

• 這樣就能進行api的調(diào)用了

• 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))

                # 循環(huán)判斷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()
                        # 獲取狀態(tài)
                        status = count_build.get_status()
                        # 獲取變更內(nèi)容
                        change = count_build.get_changeset_items()
                        log.info(" " + str(start_time) + "發(fā)起的" + my_job_name + "構建已經(jīng)完成，構建的")
                        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(" 發(fā)起的" + my_job_name + " 變更的類： " + file_list)
                                    log.info(" 發(fā)起的" + my_job_name + " 變更的備注： " + data["msg"])
                                    log.info(" 發(fā)起的" + my_job_name + " 變更的提交人： " + data["author"]["fullName"])

                            else:
                                log.info(" 發(fā)起的" + my_job_name + " 構建沒有變更內(nèi)容")

                            if len(err_list) > 0:
                                log.warning(" 構建的" + my_job_name + " 構建狀態(tài)為成功，但包含了以下錯誤：")
                                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(" 發(fā)起的" + my_job_name + " Jenkins is running")
        else:
            log.warning(" 發(fā)起的" + 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