Python实战：3分钟搞定Jenkins API批量获取Job状态（附完整代码）

Python实战3分钟搞定Jenkins API批量获取Job状态附完整代码如果你每天上班第一件事就是打开浏览器挨个点击几十个甚至上百个Jenkins Job只为了看一眼它们是不是绿的、上次构建是什么时候那么恭喜你这篇文章就是为你准备的。这种重复、低效且容易出错的监控方式早就该被自动化脚本取代了。今天我们不谈复杂的架构也不讲高深的理论就聚焦一个最实际、最高频的需求如何用Python快速、优雅地批量获取所有Jenkins Job的状态和最新构建信息。想象一下你只需要运行一个脚本就能在终端里看到一份清晰、实时的报告哪些Job最近失败了需要关注哪些已经很久没运行了哪些正在构建中。这不仅能帮你节省大量时间更能让你从繁琐的点击操作中解放出来专注于更有价值的故障排查和流程优化。无论是负责CI/CD流水线的开发工程师还是维护大量自动化任务的运维人员掌握这项技能都能让你的日常工作轻松一个量级。接下来我将从一个真实的运维视角带你一步步构建一个健壮、实用的Jenkins Job状态监控工具。1. 环境准备与核心库选择在动手写代码之前我们需要先把“厨房”收拾好。这里没有复杂的依赖核心就是两个库requests用于发起HTTP请求python-jenkins作为可选的、更友好的官方封装。我会先介绍最直接、最可控的requests方案因为它能让你更清晰地理解API的交互过程。首先确保你的Python环境已经就绪。我强烈建议使用虚拟环境来管理项目依赖避免污染全局环境。# 创建并激活虚拟环境以venv为例 python3 -m venv jenkins_monitor_env source jenkins_monitor_env/bin/activate # Linux/macOS # jenkins_monitor_env\Scripts\activate # Windows # 安装核心依赖 pip install requests如果你更喜欢更面向对象、封装度更高的方式也可以安装python-jenkins库。它内部也是基于requests但提供了更符合Python习惯的接口。pip install python-jenkins注意无论选择哪种方式你都需要拥有目标Jenkins服务器的访问权限以及一个有效的用户凭证用户名/密码或API Token。为了安全永远不要将密码硬编码在脚本中我们稍后会讨论如何安全地管理凭证。接下来我们简单对比一下两种方式的异同帮助你做出选择特性直接使用requests使用python-jenkins库控制粒度极高可以精细控制每个API请求和响应处理。中等库封装了常用操作但某些边缘情况可能需要绕开库直接调用。学习成本需要对Jenkins REST API有一定了解。较低库的接口更直观符合Pythonic风格。灵活性极强可以轻松组合、定制非标准的API调用。较强但受限于库已实现的功能。代码量相对较多需要自己处理HTTP状态码、JSON解析等。较少常用功能一行代码即可调用。适用场景需要深度定制、性能调优或调用未封装API时。快速实现标准化的Jenkins操作追求开发效率时。本文将主要以requests方案进行演示因为它能最直观地揭示原理。理解了底层逻辑后你完全可以自行迁移到python-jenkins库上。2. 构建健壮的Jenkins API客户端与Jenkins服务器对话第一步是建立连接。一个健壮的客户端不仅要能发送请求更要能妥善处理各种异常情况比如网络波动、认证失败、服务器错误等。让我们从最基础的认证开始。Jenkins最常用的认证方式是HTTP Basic Auth。你需要将用户名和API Token推荐使用Token而非登录密码作为凭证。获取Token的路径是登录Jenkins - 点击右上角用户名 - 选择“Configure” - 在“API Token”区域点击“Add new Token”。下面是一个封装了基础请求和异常处理的客户端类雏形import requests from requests.auth import HTTPBasicAuth import json import time from typing import Optional, Dict, Any, List class JenkinsAPIClient: 一个简单的Jenkins API客户端包含重试和基础错误处理。 def __init__(self, base_url: str, username: str, api_token: str): 初始化客户端。 :param base_url: Jenkins服务器地址如 http://jenkins.your-company.com:8080 :param username: Jenkins用户名 :param api_token: 用户的API Token # 确保base_url不以斜杠结尾避免后续拼接URL时出现双斜杠 self.base_url base_url.rstrip(/) self.auth HTTPBasicAuth(username, api_token) self.session requests.Session() self.session.auth self.auth # 设置一个合理的超时时间避免请求无限挂起 self.timeout 30 def _make_request(self, endpoint: str, params: Optional[Dict] None, max_retries: int 3) - Optional[Dict[str, Any]]: 内部方法发送GET请求并处理重试逻辑。 :param endpoint: API端点例如 /api/json :param params: 查询参数 :param max_retries: 最大重试次数 :return: 解析后的JSON字典失败则返回None url f{self.base_url}{endpoint} for attempt in range(max_retries): try: response self.session.get(url, paramsparams, timeoutself.timeout) response.raise_for_status() # 如果状态码不是2xx会抛出HTTPError异常 # Jenkins API通常返回JSON格式 return response.json() except requests.exceptions.ConnectionError as e: print(f[尝试 {attempt 1}/{max_retries}] 连接错误: {e}) if attempt max_retries - 1: time.sleep(2 ** attempt) # 指数退避策略 else: print(f重试{max_retries}次后仍失败请检查网络或Jenkins服务状态。) return None except requests.exceptions.HTTPError as e: # 处理特定的HTTP错误码 status_code e.response.status_code if status_code 401: print(错误认证失败请检查用户名和API Token。) elif status_code 403: print(错误权限不足该用户无权访问此资源。) elif status_code 404: print(f错误请求的资源不存在 ({url})。) else: print(fHTTP错误 {status_code}: {e}) return None except requests.exceptions.Timeout: print(f[尝试 {attempt 1}/{max_retries}] 请求超时。) if attempt max_retries - 1: time.sleep(2 ** attempt) else: print(请求超时请检查网络或Jenkins服务器负载。) return None except json.JSONDecodeError as e: print(f响应内容不是有效的JSON: {e}) return None except Exception as e: print(f发生未知错误: {e}) return None return None这个_make_request方法是我们所有后续操作的基础。它做了几件关键的事情会话复用使用requests.Session()可以保持连接并在多次请求间复用TCP连接提升效率。统一认证将认证信息挂在session上无需每次请求都传递。指数退避重试对于网络波动导致的连接错误或超时自动进行重试且等待时间逐次增加1秒2秒4秒...避免对服务器造成冲击。精细化错误处理区分了连接错误、HTTP状态码错误如401认证失败、404资源不存在、超时和JSON解析错误并给出明确的提示。有了这个稳健的“发动机”我们就可以放心地构建上层功能了。3. 核心功能实现获取所有Job及其状态现在我们来实现最核心的两个功能获取所有Job的列表以及获取每个Job的详细状态信息。Jenkins的API设计得非常清晰主要数据都通过/api/json这个入口点获取。3.1 获取所有Job名称列表首先我们获取服务器上所有Job包括文件夹内的Job的名称。这里我们调用根目录的/api/jsonAPI。def get_all_job_names(self) - List[str]: 获取Jenkins服务器上所有顶层Job的名称。 注意此方法默认只返回顶层视图下的Job。对于文件夹Folder内的Job需要递归处理。 :return: Job名称列表 data self._make_request(/api/json) if data is None: return [] job_names [] # 顶层API返回的jobs字段包含了视图和Job的信息 for item in data.get(jobs, []): # 判断是普通Job还是文件夹Folder # Jenkins中文件夹的_class通常包含Folder字样 if folder in item.get(_class, ).lower(): print(f发现文件夹: {item[name]}开始递归获取内部Job...) # 递归获取文件夹内的Job folder_jobs self._get_jobs_in_folder(item[name]) job_names.extend(folder_jobs) else: # 普通Job直接添加名称 job_names.append(item[name]) return job_names def _get_jobs_in_folder(self, folder_path: str) - List[str]: 递归获取文件夹内的所有Job名称。 :param folder_path: 文件夹路径如 my-folder 或 parent-folder/child-folder :return: 该文件夹下的所有Job名称列表包含完整路径 # 访问文件夹的API格式为 /job/{folder-name}/api/json endpoint f/job/{folder_path}/api/json data self._make_request(endpoint) if data is None: return [] jobs_in_folder [] for item in data.get(jobs, []): # 构建完整的Job路径 full_job_name f{folder_path}/{item[name]} if folder in item.get(_class, ).lower(): # 如果是嵌套文件夹继续递归 nested_jobs self._get_jobs_in_folder(full_job_name) jobs_in_folder.extend(nested_jobs) else: jobs_in_folder.append(full_job_name) return jobs_in_folder提示现代Jenkins中大量使用文件夹Folder来组织Job因此递归处理文件夹是批量操作中必不可少的一环。上述代码提供了一个简单的递归模型你可以根据实际情况调整例如设置递归深度限制。3.2 获取单个Job的详细状态获取Job名称后我们需要获取每个Job的详细状态。我们关心的信息通常包括最后构建结果SUCCESS, FAILURE, UNSTABLE, ABORTED等。最后构建时间转换为人类可读的格式。是否正在构建building字段。最后构建编号方便快速跳转到构建日志。颜色直观的视觉状态blue, red, yellow等。import datetime def get_job_details(self, job_name: str) - Optional[Dict[str, Any]]: 获取指定Job的详细状态信息。 :param job_name: Job名称包含文件夹路径如 folder/my-job :return: 包含Job详细信息的字典失败返回None # 使用 depth2 可以获取更丰富的信息包括最后构建的详情 endpoint f/job/{job_name}/api/json params {depth: 2} data self._make_request(endpoint, paramsparams) if data is None: return None details { full_name: job_name, display_name: data.get(displayName, job_name), url: data.get(url, ), color: data.get(color, notbuilt), # 颜色代表整体状态 health_report: data.get(healthReport, []), # 健康度报告 } # 处理最后构建信息 last_build data.get(lastBuild) if last_build is not None and isinstance(last_build, dict): # 转换时间戳Jenkins的时间戳是毫秒 timestamp_ms last_build.get(timestamp) if timestamp_ms: try: dt datetime.datetime.fromtimestamp(timestamp_ms / 1000) last_build_time dt.strftime(%Y-%m-%d %H:%M:%S) except (TypeError, ValueError): last_build_time N/A else: last_build_time N/A details[last_build] { number: last_build.get(number), result: last_build.get(result), # SUCCESS, FAILURE等 timestamp: last_build_time, url: last_build.get(url, ), building: last_build.get(building, False), # 是否正在构建 duration: last_build.get(duration, 0), # 持续时间毫秒 } else: # 该Job从未构建过 details[last_build] None # 获取正在进行的构建 details[in_progress_builds] [] for build in data.get(builds, []): if build.get(building): details[in_progress_builds].append({ number: build.get(number), url: build.get(url, ), }) return details这个方法返回了一个结构化的字典包含了我们需要的所有关键信息。depth参数非常有用它控制API返回数据的详细程度。depth1只返回基本信息depth2会包含最后构建的详细信息depth值越大信息越全但响应体积也越大需要根据实际情况权衡。4. 批量处理与数据呈现从脚本到实用工具有了获取单个Job详情的能力批量处理就水到渠成了。但直接循环调用可能会对服务器造成压力尤其是Job数量成百上千时。我们需要引入简单的并发控制并考虑如何将数据清晰地呈现出来。4.1 使用线程池进行并发查询为了加快批量获取的速度我们可以使用Python内置的concurrent.futures模块来并发执行多个查询任务。from concurrent.futures import ThreadPoolExecutor, as_completed def get_all_jobs_status(self, max_workers: int 10) - List[Dict[str, Any]]: 并发获取所有Job的状态详情。 :param max_workers: 线程池最大工作线程数根据Jenkins服务器性能调整。 :return: 所有Job的状态详情列表 all_job_names self.get_all_job_names() if not all_job_names: print(未获取到任何Job名称。) return [] print(f开始获取 {len(all_job_names)} 个Job的状态信息...) all_job_details [] # 使用线程池并发执行 with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_job {executor.submit(self.get_job_details, job_name): job_name for job_name in all_job_names} for future in as_completed(future_to_job): job_name future_to_job[future] try: details future.result() if details: all_job_details.append(details) else: print(f警告未能获取Job {job_name} 的详情。) except Exception as e: print(f获取Job {job_name} 详情时发生异常: {e}) print(f状态获取完成成功获取 {len(all_job_details)} 个Job的信息。) return all_job_details这里将max_workers默认设置为10。这个数字不宜过大否则可能会对Jenkins服务器造成瞬间高并发请求导致服务响应变慢甚至拒绝服务。建议根据实际情况服务器性能、网络状况、Job数量进行调整。4.2 数据格式化与输出获取到数据后我们需要以一种清晰、易读的方式呈现。终端表格是一个不错的选择。我们可以使用tabulate库来美化输出。pip install tabulate然后编写一个格式化输出的函数from tabulate import tabulate def print_jobs_table(job_details_list: List[Dict[str, Any]]): 将Job详情列表以表格形式打印到控制台。 :param job_details_list: get_all_jobs_status返回的列表 if not job_details_list: print(没有可展示的Job数据。) return table_data [] for job in job_details_list: last_build job.get(last_build) if last_build: build_status last_build.get(result, N/A) build_time last_build.get(timestamp, N/A) building 是 if last_build.get(building) else 否 else: build_status 从未构建 build_time N/A building 否 # 根据颜色或状态添加简单的状态标识 color job.get(color, ) status_icon if blue in color: status_icon ✅ elif red in color: status_icon ❌ elif yellow in color: status_icon ⚠️ else: status_icon table_data.append([ status_icon, job.get(display_name), build_status, build_time, building, job.get(url) ]) headers [状态, Job名称, 最后结果, 最后构建时间, 正在构建, URL] # 使用grid格式在终端显示效果较好 print(tabulate(table_data, headersheaders, tablefmtgrid, maxcolwidths[None, 30, None, 20, None, 50]))这个表格会清晰地展示每个Job的名称、最后构建结果、时间、是否正在运行以及链接。状态图标提供了快速的视觉反馈。4.3 进阶生成HTML报告对于需要存档或通过邮件发送的场景一个HTML报告会更专业。我们可以用Jinja2模板来生成。pip install Jinja2创建一个简单的模板文件report_template.html!DOCTYPE html html head titleJenkins Job状态报告 - {{ timestamp }}/title style body { font-family: sans-serif; margin: 20px; } table { border-collapse: collapse; width: 100%; } th, td { border: 1px solid #ddd; padding: 8px; text-align: left; } th { background-color: #f2f2f2; } .success { background-color: #d4edda; } .failure { background-color: #f8d7da; } .unstable { background-color: #fff3cd; } .building { font-weight: bold; color: #856404; } a { color: #007bff; text-decoration: none; } a:hover { text-decoration: underline; } /style /head body h1Jenkins Job状态报告/h1 p生成时间: {{ timestamp }} | 总计: {{ total_jobs }} 个Job/p table thead tr thJob名称/th th最后结果/th th最后构建时间/th th状态/th th链接/th /tr /thead tbody {% for job in jobs %} tr class{{ building if job.last_build and job.last_build.building else job.last_build.result|lower if job.last_build else }} td{{ job.display_name }}/td td {% if job.last_build %} {{ job.last_build.result }} {% else %} 从未构建 {% endif %} /td td {% if job.last_build %} {{ job.last_build.timestamp }} {% else %} N/A {% endif %} /td td {% if job.last_build and job.last_build.building %} span classbuilding 构建中 (#{{ job.last_build.number }})/span {% elif job.last_build %} {% if job.last_build.result SUCCESS %} span stylecolor:green;✅ 成功/span {% elif job.last_build.result FAILURE %} span stylecolor:red;❌ 失败/span {% elif job.last_build.result UNSTABLE %} span stylecolor:orange;⚠️ 不稳定/span {% else %} {{ job.last_build.result }} {% endif %} {% else %} ⏸️ 未构建 {% endif %} /td tda href{{ job.url }} target_blank查看/a/td /tr {% endfor %} /tbody /table /body /html然后在代码中渲染并保存报告from jinja2 import Environment, FileSystemLoader import datetime def generate_html_report(job_details_list: List[Dict[str, Any]], output_path: str jenkins_report.html): 生成HTML格式的状态报告。 :param job_details_list: Job详情列表 :param output_path: HTML报告输出路径 env Environment(loaderFileSystemLoader(.)) # 假设模板在当前目录 template env.get_template(report_template.html) timestamp datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) html_content template.render( timestamptimestamp, total_jobslen(job_details_list), jobsjob_details_list ) with open(output_path, w, encodingutf-8) as f: f.write(html_content) print(fHTML报告已生成: {output_path})5. 实战整合与安全增强现在我们把所有部分组合起来形成一个完整的、可执行的脚本。同时我们还要解决一个至关重要的问题凭证的安全管理。5.1 完整的脚本示例创建一个文件例如jenkins_job_monitor.py#!/usr/bin/env python3 Jenkins Job状态批量监控脚本。 安全地从环境变量或配置文件中读取凭证并发获取所有Job状态并输出表格和HTML报告。 import os import sys import argparse from typing import Optional # 导入我们之前编写的类和方法 # 假设 JenkinsAPIClient, print_jobs_table, generate_html_report 等都在同一个文件或已导入 def main(): parser argparse.ArgumentParser(description批量获取Jenkins Job状态信息) parser.add_argument(--url, requiredTrue, helpJenkins服务器基础URL如 http://jenkins.example.com:8080) parser.add_argument(--user, helpJenkins用户名也可通过环境变量JENKINS_USER设置) parser.add_argument(--token, helpJenkins API Token也可通过环境变量JENKINS_TOKEN设置) parser.add_argument(--output, defaultjenkins_report.html, helpHTML报告输出路径默认: jenkins_report.html) parser.add_argument(--workers, typeint, default10, help并发查询线程数默认: 10) args parser.parse_args() # 1. 安全地获取凭证优先级命令行参数 环境变量 username args.user or os.environ.get(JENKINS_USER) api_token args.token or os.environ.get(JENKINS_TOKEN) if not username or not api_token: print(错误未提供认证凭证。) print(请通过以下方式之一提供) print( 1. 命令行参数: --user 用户名 --token API_TOKEN) print( 2. 环境变量: 设置 JENKINS_USER 和 JENKINS_TOKEN) sys.exit(1) # 2. 初始化客户端 print(f正在连接至 Jenkins: {args.url}) client JenkinsAPIClient(base_urlargs.url, usernameusername, api_tokenapi_token) # 3. 获取所有Job状态 all_job_details client.get_all_jobs_status(max_workersargs.workers) if not all_job_details: print(未获取到任何Job状态信息程序退出。) sys.exit(0) # 4. 控制台输出表格 print(\n *80) print(Jenkins Job状态概览) print(*80) print_jobs_table(all_job_details) # 5. 生成HTML报告 try: generate_html_report(all_job_details, args.output) except Exception as e: print(f生成HTML报告时出错: {e}) # 6. 可选简单统计 success_count sum(1 for j in all_job_details if j.get(last_build) and j[last_build].get(result) SUCCESS) failure_count sum(1 for j in all_job_details if j.get(last_build) and j[last_build].get(result) FAILURE) building_count sum(1 for j in all_job_details if j.get(last_build) and j[last_build].get(building)) print(f\n统计: 成功 {success_count} | 失败 {failure_count} | 构建中 {building_count} | 总计 {len(all_job_details)}) if __name__ __main__: main()5.2 安全使用凭证的最佳实践将密码或Token写在脚本里是极其危险的。推荐以下几种安全实践环境变量如上例所示通过环境变量传递。可以在shell配置文件中设置export JENKINS_USERyour_username export JENKINS_TOKENyour_api_token_here然后在脚本中通过os.environ.get(JENKINS_USER)读取。配置文件使用配置文件如YAML、JSON或.ini格式并确保该文件有严格的权限如chmod 600 config.yaml。脚本读取配置文件获取凭证。密钥管理服务在生产环境中考虑使用如HashiCorp Vault、AWS Secrets Manager或Azure Key Vault等服务来动态获取凭证。使用keyring库桌面环境对于个人开发机可以使用Python的keyring库将凭证安全地存储在系统的密钥环中。5.3 运行脚本现在你可以这样运行脚本了# 方式1通过环境变量推荐 export JENKINS_USERalice export JENKINS_TOKEN114abc123def456... python jenkins_job_monitor.py --url http://your-jenkins-server:8080 # 方式2通过命令行参数不推荐因为Token会出现在进程列表和shell历史中 python jenkins_job_monitor.py --url http://your-jenkins-server:8080 --user alice --token 114abc123def456... # 指定更多工作线程和输出路径 python jenkins_job_monitor.py --url http://your-jenkins-server:8080 --workers 15 --output ./reports/daily_status.html运行后你会在终端看到一个清晰的表格并在当前目录得到一个名为jenkins_report.html的详细报告可以直接用浏览器打开。6. 扩展思路从监控到自动化掌握了批量获取状态的能力后你可以轻松地将这个脚本扩展成一个功能更强大的自动化工具。这里有几个我实践过的方向定时任务与报警结合cronLinux或Task SchedulerWindows让脚本定时运行。当发现失败Job或长时间未运行的Job时通过邮件smtplib、Slackslack_sdk或钉钉/webhook发送报警通知。数据持久化与趋势分析将每次获取的状态信息存入数据库如SQLite、PostgreSQL或时间序列数据库如InfluxDB。这样你就可以分析构建成功率的历史趋势、识别不稳定的Job、统计平均构建时间等为优化CI/CD流水线提供数据支持。自动修复与重试对于一些已知的、可自动恢复的失败场景例如因依赖服务临时不可用导致的失败可以在脚本中集成自动重试逻辑。检测到特定错误模式的失败Job后自动触发一次新的构建。生成仪表盘将生成的HTML报告部署到一个内部Web服务器上或者使用Grafana等可视化工具连接到你存储的历史数据创建一个实时更新的Jenkins健康状态仪表盘。这个脚本的核心价值在于它把一个手动、重复、易错的操作变成了一个可靠、高效、可扩展的自动化流程。我自己的团队已经把它集成到了每日的晨会报告中节省了大家至少15分钟的手动检查时间。更重要的是它让我们能更早地发现潜在问题而不是等到有人手动点开才发现某个关键流水线已经失败了好几天。