Python Jenkins API

Python Jenkins API

作者：运维人在路上

个人博客：https://www.cnblogs.com/hujinzhong

微信公众号：运维人在路上

Bilibili账号：https://space.bilibili.com/409609392

个人简介：本人是一枚大型电商平台的运维工程师，对开发及运维有一定了解，现分享技术干货，欢迎大家交流！

一、python-jenkins模块

1.1、安装

pip install python-jenkins

1.2、相关文档

# 官方文档：
	https://python-jenkins.readthedocs.io/en/latest/
# 官方API：
	https://python-jenkins.readthedocs.io/en/latest/api.html

1.3、相关方法

import jenkins

server = jenkins.Jenkins('http://10.0.0.102:8888/', username="admin", password="123456")

# Jenkins的job数量
print(server.jobs_count())

# Jenkins-web-ui
print(server.server)    # http://10.0.0.102:8888/

# 所有的job信息
all_jobs_li = server.get_all_jobs()
for item in all_jobs_li:
    print('name: %s' % item['name'], 'URL: ', item['url'])

# 账户信息描述
print(server.get_whoami())

# auth
print(server.auth)

# DEBUG Job信息
print(server.debug_job_info('web_demo_freestyle'))

# ####### CURD Job #######
# Job是否存在
print(server.job_exists('web_demo_freestyle'))

# 获取所有的Job
print(server.get_jobs())

# 创建Job
server.create_job('API', jenkins.RECONFIG_XML)

# 删除Job
server.delete_job('API')

# 复制job
server.copy_job('web_demo_freestyle', 'web_demo_freestyle_copy')

# enable job
server.enable_job('web_demo_freestyle_copy')

# 禁用Job
server.disable_job('copy-my-github')

# 重新配置Job
server.reconfig_job('copy-my-github', jenkins.RECONFIG_XML)

# 重命名Job
server.rename_job('web_demo_freestyle_copy_2', 'web_demo_freestyle_copy')

# 触发Job运行
# 触发Job(方式一)
server.build_job_url('web_demo_freestyle')

# 触发Job(方式二)
server.build_job('web_demo_freestyle')
# server.build_job('web_demo_freestyle_copy', parameters=params)

# 查看指定构建编号的输出
print(server.get_build_console_output('web_demo_freestyle', 11))

# 下一次构建编号,步长为5
next_bn = server.get_job_info('ansible-playbook')['nextBuildNumber']
server.set_next_build_number('ansible-playbook', next_bn + 5)

# 指定编号的构建Job信息
print(server.get_build_info('web_demo_freestyle', 11))

# 获取正在运行的Job，一般结合build_job方法一起使用
print(server.get_running_builds())

# 终止指定编号的Job
server.stop_build('ansible-playbook', 62)

# 获取Job名称
print(server.get_job_name('web_demo_freestyle'))

# 断言Job是否存在
print(server.assert_job_exists('web_demo_freestyle'))
'''
存在则返回None，不存在则抛出错误信息
'''

# 该Job所有的相关信息
print(server.get_job_info('web_demo_freestyle'))
print(server.get_job_info('web_demo_freestyle')['actions'])

# 查看该视图下所有的Job
print(server.get_jobs())
print(server.get_jobs(view_name='all'))

# 正则匹配获取Job信息
print(server.get_job_info_regex('^dianchou'))

# 获取最后一次构建的Job信息
job_name = "web_demo_freestyle"
server.build_job(job_name)
last_build_number = server.get_job_info(job_name)['lastCompletedBuild']['number']
build_info = server.get_build_info(job_name, last_build_number)
print(build_info)

# shutdown jenkins
# server.quiet_down()

1.4、脚本分享

#!usrinpython3
#  -*- coding: utf-8 -*-
import logging

import requests

try:
    import jenkins
except ImportError:
    jenkins = None
    print("未安装Jenkins模块
pip install python-jenkins")

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


def build_package():
    """启动依赖包构建任务"""
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        if 'package' in job_name:
            result = server.build_job(job_name)
            logger.info('build ' + job_name + ' ' + str(result))


def build_all_job():
    """启动所有非package任务"""
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        if 'package' not in job_name:
            result = server.build_job(job_name)
            logger.info('build ' + job_name + ' ' + str(result))


def build_all_docker():
    """启动所有非package任务"""
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        if 'auto-test_83,' not in job_name:
            continue
        if 'boss' in job_name:
            continue
        if 'package' not in job_name:
            result = server.build_job(job_name)
            logger.info('build ' + job_name + ' ' + str(result))


def build_all_dubbo_job():
    """启动所有非package任务"""
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        if 'auto-' in job_name and 'package' not in job_name:
            try:
                result = server.build_job(job_name)
                logger.info('build ' + job_name + ' ' + str(result))
            except requests.exceptions.HTTPError:
                logger.info('build ' + job_name + ' ' + 'requests.exceptions.HTTPError')


def build_all_cloud_job():
    """启动所有非package任务"""
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        if 'auto2-' in job_name and 'package' not in job_name:
            try:
                result = server.build_job(job_name)
                logger.info('build ' + job_name + ' ' + str(result))
            except requests.exceptions.HTTPError:
                logger.info('build ' + job_name + ' ' + 'requests.exceptions.HTTPError')


def stop_all():
    jobs = server.get_all_jobs()
    for job in jobs:
        job_name = job['name']
        builds = server.get_job_info(job_name)['builds']
        for build in builds:
            try:
                server.stop_build(job_name, build['number'])
                print('stop', job_name, build['number'])
            except jenkins.NotFoundException:
                pass


if __name__ == '__main__':
    server = jenkins.Jenkins('http://10.0.0.102:8888', username="admin", password="123456")
    stop_all()
    # build_package()
    # build_all_cloud_job()
    # build_all_dubbo_job()