在持续集成:Jenkins API简单使用中介绍了使用Python requests 和 jenkinsapi 库来请求Jenkins API。本文将更全面的介绍Jenkins API以及使用方法。
目录Jenkins API简介获取信息
XML APIJSON API请求信息
curl请求Python requests 触发构建
无参数构建参数化构建 删除操作Python API
Jenkins API简介我们可以使用jenkins API来获取jenkins平台相关信息、触发构建、创建/删除job等,这些API使得jenkins具备了很强的扩展性,比如集成其它平台。
Jenkins API采用的是REST架构设计风格,支持以下3种方式:
XML API:.../api/xmlJSON API:.../api/jsonPython API:比如有 JenkinsAPI, Python-Jenkins, api4jenkins, aiojenkins Python库。
Jenkins的很多页面都提供了远程访问API,你可以在当前页面URL后添加 /api 查看当前接口说明。
除了在服务器上手动重启jenkins外,还可以通过API进行重启(POST请求):
重启: 以下面这个页面为例: 在URL后面添加 api 返回如下页面: 在URL后添加/api/xml,效果如下: 下面介绍过滤xml信息的方法。 1、使用xpath语法 读取测试结果:.../api/xml?xpath=action[4] 读取用例总数totalCount:/api/xml?xpath=action 3、depth指定深度 指定深度为1:/api/xml?depth=1 4、tree参数 可以使用tree参数来提取以及组合指定字段:/api/xml?pretty=true&tree=building,displayName,id,url,previousBuild[url] 如果返回多组结果,可以指定范围(索引从0开始): {M,N}: 从第M个元素到第N个元素{M,}: 从第M个元素到最后一个元素{,N}: 从第1个元素到第N个元素,等价于 {0,N}{N}: 第N个元素,等价于 {N,N+1}
比如下面的页面: 获取第1个到第2个元素:.../api/xml?tree=jobs[name]{1,3} 5、组合 可以使用 & 来组合多个表达式,比如:/api/xml?exclude=/*/action&xpath=//fullDisplayName 6、wrapper方法整合多个结果 过滤后可能有多个结果,需要使用wrapper方法来组合,例如:/api/xml?depth=2&xpath=//fullDisplayName&wrapper=jobname JSON API返回json样式的数据,在URL后添加/api/json。它也支持depth参数,默认为最小深度。 来看一下效果:...3/api/json?pretty=true JSON API也可以使用tree参数来过滤:/api/json?pretty=true&tree=building,displayName,id,url,previousBuild[url] 由于jenkins API采用的是REST架构风格,所以支持REST API的请求方法,比如常用的POST、GET、DELETe等方法。 获取job的构建Number的API: 获取最近的job buildNumber: 介绍两种方式请求api,第一种方法是curl工具,第二种是Python的requests库。 使用如下curl命令发送GET请求: API Token可以在用户配置中生成。 还是以下面这个页面为例: 1、解析xml结果 使用xpath过滤robot测试结果:/api/xml?xpath=//action[@_class='hudson.plugins.robot.RobotBuildAction'] shell脚本中,可以对返回的信息进行进一步处理,比如,读取totalCount: 读取 failCount 和 totalCount: 在需要读取多个值的时候,可以一个一个的请求,我的习惯是只请求一次,然后进行二次处理,比如在groovy脚本中: 2、解析json结果 获取json格式数据和xml类似,json数据的进一步提取可以结合jq命令来使用。这里介绍一个在groovy中处理json数据的方法。 groovy脚本中可以使用readJSON 方法将json字符串反序列化为json对象,然后提取对应的值。 注意:在jenkins pipeline中使用readJSON 方法需要安装插件 Pipeline Utility Steps Plugin 。 代码如下: 介绍第二种请求REST API的工具Python requests库,举几个例子: 获取build number: 读取failCount: 查看当前job是否正在构建: 通常使用POST请求在触发构建Job,请求地址为: 1、curl命令 2、Python requests库 对RF-Pipeline-Demo进行参数化构建配置,添加两个字符串参数param1和param2。 注意:参数化构建功能需要安装插件 Parameterized Trigger。 1、curl命令 2、Python requests库 删除操作使用POST请求: 删除Job某次构建: 禁用和启用Job: 禁用指定job: 前面已经介绍了常见的Jenkins API以及使用方法,Python、Ruby和Java都对jenkins API 进行了封装,使我们使用起来更加方便。下面简单介绍Python JenkinsAPI库的使用方法。 可以在URL后添加 /api/python 获取json格式数据:/api/python?pretty=true&tree=building,displayName,id,url,previousBuild[url] JenkinsAPI, Python-Jenkins, api4jenkins, aiojenkins等Python库封装了许多jenkins REST API,下面以JenkinsAPI库为例。 直接来看代码: 执行结果: 上述代码仅展示了部分方法,还可以创建、删除、复制job,创建、删除节点,重启等。 更多方法可以查看库的相关函数。 上德不德,是以有德;下德不失德,是以无德。——《道德经》$ curl -k --silent -L --user 用户名:API Token JENKINS_URL/job/JOB_NAME
$ curl -k --silent -L --user admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/view/demo/job/RF-Pipeline-Demo/3/api/xml?xpath=//action[@_class='hudson.plugins.robot.RobotBuildAction'] | tr '<' 'n' | egrep '^totalCount>' | sed 's/.*>//g'
4
$
$ curl -k --silent -L --user admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/view/demo/job/RF-Pipeline-Demo/3/api/xml?xpath=//action[@_class='hudson.plugins.robot.RobotBuildAction'] | tr '<' 'n' | egrep '^totalCount>|^failCount>' | sed 's/>/:/g' | sed -e '1s/$/,/g' | tr -d 'n'
failCount:0,totalCount:4
// curl读取测试结果
def WORKFLOW_XML = sh ( script: "API_XML=$(curl -k --silent -L --user admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/view/demo/job/RF-Pipeline-Demo/3/api/xml?xpath=//action[@_class='hudson.plugins.robot.RobotBuildAction'] | tr '<' 'n' | egrep '^totalCount>|^failCount>' | sed 's/>/:/g' | sed -e '1s/$/,/g' | tr -d 'n'); echo ${API_XML}", returnStdout: true ).trim();
rf_results = WORKFLOW_XML.split(",");
if (rf_results[0]) {
for ( res in rf_results ) {
if (res.contains("failCount")) {
failCount = res.split("failCount:")[1];
}
if (res.contains("totalCount")) {
totalCount = res.split("totalCount:")[1];
}
}
passCount = totalCount.toInteger() - failCount.toInteger();
passPercentage = passCount/totalCount.toInteger() * 100;
passPercentage = Math.round(passPercentage * 100) / 100; // 保留2位小数
passRatio = passCount + " / " + totalCount;
} else {
passPercentage = 'null';
passRatio = 'null';
}
def WORKFLOW_JSON = sh ( script: "API_JSON=$(curl -k --silent -L --user admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/view/demo/job/RF-Pipeline-Demo/3/api/json); echo ${API_JSON}", returnStdout: true ).trim()
def jsonObj = readJSON text: WORKFLOW_JSON;
def failCount = (jsonObj.actions.failCount - null)[0]; // 去掉list中的null
def totalCount = (jsonObj.actions.totalCount - null)[0];
Python requests
import requests
url = "http://192.168.30.8:8080/job/RF-Pipeline-Demo/lastBuild/buildNumber"
ret = requests.get(url)
print(ret.text)
import json
import requests
url = "http://192.168.30.8:8080/view/demo/job/RF-Pipeline-Demo/3/api/json"
ret = requests.get(url)
res = ret.json()
print(res["actions"][3]["failCount"])
print(json.dumps(ret.json(),indent=2))
def checkBuilding(self,jobname):
# 查看当前job是否正在构建
url = "http://192.168.30.8:8080/job/{jobname}/api/json?tree=lastCompletedBuild[number],lastBuild[number]"
res = requests.get(url)
res = res.json()
if res["lastBuild"]["number"] == res["lastCompletedBuild"]["number"]:
return False
else:
print(f"{jobname}正在构建中...")
return True
触发构建
$ curl -k --silent -L -X POST --USER admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/job/RF-Pipeline-Demo/build
import requests
url = "http://192.168.30.8:8080/job/RF-Pipeline-Demo/build"
username = "admin"
api_token = "11f8790b23a9983c0a218ba125aa855f61"
res = requests.post(url, auth=(username, api_token), verify=False)
参数化构建
# 字符参数
$ curl -k --silent -L -X POST --data param1=1 --data param2=2 --user admin:11f8790b23a9983c0a218ba125aa855f61 http://192.168.30.8:8080/job/RF-Pipeline-Demo/buildWithParameters
$
import requests
jenkins_params = {'param1': 'value11',
'param2': 'value22'}
url = "http://192.168.30.8:8080/job/RF-Pipeline-Demo/buildWithParameters"
username = "admin"
api_token = "11f8790b23a9983c0a218ba125aa855f61"
res = requests.post(url, auth=(username, api_token), params=jenkins_params, verify=False)
删除操作
import datetime
from jenkinsapi.jenkins import Jenkins
class JenkinsApiDemo:
def __init__(self, job_name, chose='jenkins'):
self.jenkins_host = '192.168.30.8'
self.jenkins_port = '8080'
self.username = 'admin'
self.pwd = 'admin'
self.jenkins_url = f"http://{self.jenkins_host}:{self.jenkins_port}"
self.job_name = job_name
self.jenkins_server = Jenkins(self.jenkins_url, username=self.username, password=self.pwd, useCrumb=True)
def GetJenkinsVersion(self):
print(self.jenkins_server.version)
return self.jenkins_server.version
def jobBuild(self):
if self.jenkins_server.has_job(self.job_name):
myjob = self.jenkins_server.get_job(self.job_name)
if not myjob.is_queued_or_running():
self.jenkins_server.build_job(self.job_name)
def disableJob(self):
"""Disable a Jenkins job"""
if self.jenkins_server.has_job(self.job_name):
job_instance = self.jenkins_server.get_job(self.job_name)
job_instance.disable()
print('Job %s Is Enabled ?:%s' % (self.job_name, job_instance.is_enabled()))
def enableJob(self):
"""Disable a Jenkins job"""
if self.jenkins_server.has_job(self.job_name):
job_instance = self.jenkins_server.get_job(self.job_name)
job_instance.enable()
print('Job %s Is Enabled ?:%s' % (self.job_name, job_instance.is_enabled()))
def getJobInfo(self):
if self.jenkins_server.has_job(self.job_name):
myjob = self.jenkins_server.get_job(self.job_name)
if not myjob.is_queued_or_running():
last_buildnumber = myjob.get_last_buildnumber()
print("last_buildnumber: ", last_buildnumber)
last_build = myjob.get_build(last_buildnumber)
# 获取开始时间
start_time = last_build.get_timestamp() + datetime.timedelta(hours=8)
print("start_time: ", start_time)
print("status: ", last_build.get_status())
print("build_url: ", last_build.get_build_url())
print("duration: ", last_build.get_duration())
print("causes: ", last_build.get_causes()[0]["shortDescription"])
print("change: ", last_build.get_changeset_items())
print("console_log: ", last_build.get_console())
else:
print(self.job_name + " is running")
else:
print("没有 " + self.job_name + " 这个job")
if __name__ == '__main__':
jobname = "RF-Pipeline-Demo"
jk = JenkinsApiDemo(jobname)
jk.GetJenkinsVersion()
# jk.jobBuild()
jk.disableJob()
jk.enableJob()
jk.getJobInfo()
2.326
last_buildnumber: 20
start_time: 2022-01-20 18:22:12+00:00
status: SUCCESS
build_url: http://192.168.30.8:8080/job/RF-Pipeline-Demo/20/
duration: 0:00:02.620000
causes: Started by user admin
change: []
console_log: Started by user admin
[Pipeline] Start of Pipeline
[Pipeline] node
.............
jenkins_server.create_job(jobname, xml)
jenkins_server.copy_job(jobname, newjobname)
jenkins_server.create_node(name)
jenkins_server.delete_job(jobname)
jenkins_server.delete_node(nodename)
jenkins_server.safe_restart()
jenkins_server.shutdown()



