Python Jenkins-調用_get_view_jobs方法獲取指定視圖下的Job名稱

一、python jenkins庫 api簡介

1.1、連接Jenkins服務器

server = jenkins.Jenkins(url, username=None, password=None, timeout=30)

• url : 指的是所連接的Jenkins的網絡地址
• username/password : username/password 默認值是None， 在內網中不傳參數也是可以連接成功的，但應該只有讀的權限，沒有操作權限。所以一般需要用戶名和密碼。

1.3、賬戶信息

• 獲取連接賬戶的相關信息， 有關私密信息就不打印出來了

server.get_whoami()

• 獲取連接Jenkins的版本信息

server.get_version()
'2.176.3'

1.4、獲取job相關信息

• 所謂job的話，指的是在Jenkins里面創建的一系列項目

# 獲取所有的job信息
server.get_jobs()
 
# 獲取job為name的基本信息
server.get_job_info(name)
 
#獲取job名為name的job的最后次構建號， 不會計算排隊中的構建， 
#例如正在構建的number是3， 在排隊的有4和5， 那么最后次構建號是3
server.get_job_info(name)['lastBuild']['number']
 
# 正則匹配到相應job的基本信息
server.get_job_info_regex(pattern)
 
#判斷job是否存在
server.job_exists(name)
 
#獲取所有job的數量
server.jobs_count()

1.5、獲取build相關信息

• 所謂build的話，指的是在Job里面的一次構建

# 獲取某次構建信息，舉例178， 如果178不存在或者正在排隊中， 則會拋出異常
# JenkinsException: job[name] number[178] does not exist
server.get_build_info(name, 178)
 
# result指構建運行的狀態， 構建如果是運行完成返回結果FAILURE/ABORTED/SUCCESS/UNSTABLE等
# 構建如果是正在運行返回結果 None
server.get_build_info(name, 178)['result']

1.6、操作

• 需要明確該賬戶在Jenkins中有對應的操作權限，否則在腳本中是無法做相應操作的

• 獲取所有排隊的build

server.get_queue_info()

• 獲取所有正在運行的build

server.get_running_builds()
server.stop_build(name, number) # 停止正在運行的構建

• 停止掉queue排隊中的構建， 直接拋出異常NotFoundException: Requested item could not be found
• 停止掉running運行中的構建， 無返回值， 無報錯，在Jenkins頁面上顯示該構建被XXX 賬戶停掉了
• 停止掉stopped已經停止的構建， 無返回值， 無報錯，Jenkins頁面無變化

server.cancel_queue(id) # 配合server.get_queue_info()使用，得到排隊構建的id， 可以取消該構建

• server.get_queue_info() 獲取某個排隊構建的id， 如123.
• 如果該構建排隊中直接server.cancel_queue(123)， 無返回值， 無報錯，Jenkins頁面上排隊的構建直接被撤銷，看不出原來有這么一個排隊的構建
• 如果等排隊的構建到運行狀態， 使用server.cancel_queue(123)， 無返回值， 無報錯， 該構建依舊在運行中
• 如果等排隊的構建到完成狀態， 使用server.cancel_queue(123)， 無返回值， 無報錯， 該構建依舊為完成結束狀態

二、_get_view_jobs方法

from flask import Flask
from flask import request, jsonify, abort
import jenkins
import json
 
 
class Jenkins_Api:
    def __init__(self, username='賬號', password='密碼'):
        self._url = 'https://jenkins-rd.Jenkins Daemon.com/'
        self._username = username
        self._password = password
 
    def get_server(self):
        server = jenkins.Jenkins(self._url, username=self._username, password=self._password)
        return server
 
    def get_version(self):
        get_version = self.get_server().get_version()
        print(get_version)
        return get_version  # .get_version()
 
    def get_jobs(self):
        get_jobs = self.get_server().get_jobs()
        print(get_jobs)
        return get_jobs
 
    def get_views(self):
        get_views = self.get_server().get_views()
        print(get_views)
        return get_views
 
    def get_all_jobs(self):
        get_all_jobs = self.get_server().get_all_jobs()
        print(get_all_jobs)
        return get_all_jobs
 
    # _get_view_jobs(name='Prod')
    def get_views_job(self, name):
        get_views_job = self.get_server()._get_view_jobs(name=name)
        # print(get_views_job)
        job_list = []
        for job in get_views_job:
            jobs = job['name'][:-10]
            job_list.append({"application":jobs,})
            # print(job['name'].strip('Prod-CD-job'))
            # print(job_list)
 
        response = {
            "message": "",
            "code": 0,
            "data": job_list,
            "result": "true",
            "request_id": "975184016b50440a8c195040fc1443a5"
        }
 
        print(response)
        return response
 
 
app = Flask(__name__)
 
 
@app.route('/bk_itms/prod/jobs', methods=['GET', 'POST'])
def webhook():
    if request.method == 'GET':
        # logging.debug('Get Method not implemented')
        payload = JK.get_views_job(name="RC")
        return jsonify(payload), 200
    elif request.method == 'POST':
        return jsonify({'status': 'success'}), 200
 
    else:
        abort(400)
 
 
if __name__ == '__main__':
    JK = Jenkins_Api()
 
    app.run(host='0.0.0.0', port='10010')
    # print(jenkins.__dict__)