From 0789d41674547cd30d82eb530d51308f456cbe90 Mon Sep 17 00:00:00 2001 From: tech Date: Mon, 9 Mar 2026 15:32:48 +0800 Subject: [PATCH] Add 'running' command to list currently running Jenkins jobs - Add new 'running' CLI command with optional --count flag - Implement list_running_jobs method in JenkinsClient - Apply code formatting for improved readability Co-Authored-By: Claude Opus 4.6 --- src/zzpyjenkins/cli.py | 38 +++++++++-- src/zzpyjenkins/jenkins_client.py | 107 ++++++++++++++++++++++++++---- 2 files changed, 127 insertions(+), 18 deletions(-) diff --git a/src/zzpyjenkins/cli.py b/src/zzpyjenkins/cli.py index 79b6579..58f86c1 100644 --- a/src/zzpyjenkins/cli.py +++ b/src/zzpyjenkins/cli.py @@ -13,11 +13,20 @@ console = Console() @click.group() @click.option("--url", "-u", help="Jenkins server URL (default: JENKINS_URL env)") -@click.option("--username", "-U", help="Jenkins username (default: JENKINS_USERNAME env)") -@click.option("--password", "-P", help="Jenkins password/token (default: JENKINS_PASSWORD env)") +@click.option( + "--username", "-U", help="Jenkins username (default: JENKINS_USERNAME env)" +) +@click.option( + "--password", "-P", help="Jenkins password/token (default: JENKINS_PASSWORD env)" +) @click.version_option() @click.pass_context -def main(ctx: click.Context, url: Optional[str], username: Optional[str], password: Optional[str]): +def main( + ctx: click.Context, + url: Optional[str], + username: Optional[str], + password: Optional[str], +): """A CLI tool to interact with Jenkins server for building jobs.""" ctx.ensure_object(dict) ctx.obj["url"] = url @@ -29,11 +38,17 @@ def get_client(ctx: click.Context) -> JenkinsClient: """Create Jenkins client from options or environment variables.""" url = ctx.obj.get("url") or os.environ.get("JENKINS_URL") username = ctx.obj.get("username") or os.environ.get("JENKINS_USERNAME") - password = ctx.obj.get("password") or os.environ.get("JENKINS_PASSWORD") or os.environ.get("JENKINS_TOKEN") + password = ( + ctx.obj.get("password") + or os.environ.get("JENKINS_PASSWORD") + or os.environ.get("JENKINS_TOKEN") + ) if not all([url, username, password]): console.print("[red]Error: Missing Jenkins configuration.[/red]") - console.print("Use --url/--username/--password options or set environment variables:") + console.print( + "Use --url/--username/--password options or set environment variables:" + ) console.print(" JENKINS_URL, JENKINS_USERNAME, JENKINS_PASSWORD") raise SystemExit(1) @@ -59,7 +74,9 @@ def list_jobs(ctx: click.Context, pattern: Optional[str]): @main.command() @click.argument("job_name") -@click.option("--params", "-p", multiple=True, help="Build parameters in KEY=VALUE format") +@click.option( + "--params", "-p", multiple=True, help="Build parameters in KEY=VALUE format" +) @click.pass_context def build(ctx: click.Context, job_name: str, params: tuple[str, ...]): """Trigger a build for a job.""" @@ -94,5 +111,14 @@ def watch(ctx: click.Context, job_name: str): client.watch_build(job_name) +@main.command() +@click.option("--count", "-c", is_flag=True, help="Only show count of running jobs") +@click.pass_context +def running(ctx: click.Context, count: bool): + """List all currently running jobs.""" + client = get_client(ctx) + client.list_running_jobs(count_only=count) + + if __name__ == "__main__": main() diff --git a/src/zzpyjenkins/jenkins_client.py b/src/zzpyjenkins/jenkins_client.py index 4bcb60e..8076e43 100644 --- a/src/zzpyjenkins/jenkins_client.py +++ b/src/zzpyjenkins/jenkins_client.py @@ -1,7 +1,7 @@ """Jenkins client wrapper with rich output support.""" import time -from typing import Any, Optional +from typing import Optional import jenkins from rich.console import Console @@ -25,10 +25,12 @@ class JenkinsClient: version = self.server.get_version() jobs = self.server.get_all_jobs() - console.print(f"\n[bold]Jenkins Server Information[/bold]\n") + console.print("\n[bold]Jenkins Server Information[/bold]\n") console.print(f" URL: {self.url}") console.print(f" Version: {version}") - console.print(f" User: {user.get('fullName', user.get('id', 'Unknown'))}") + console.print( + f" User: {user.get('fullName', user.get('id', 'Unknown'))}" + ) console.print(f" Jobs: {len(jobs)}") except jenkins.JenkinsException as e: console.print(f"[red]Error connecting to Jenkins: {e}[/red]") @@ -63,7 +65,9 @@ class JenkinsClient: if parameters: queue_id = self.server.build_job(job_name, parameters=parameters) - console.print(f"Build #{next_build} queued with parameters: {parameters}") + console.print( + f"Build #{next_build} queued with parameters: {parameters}" + ) else: queue_id = self.server.build_job(job_name) console.print(f"Build #{next_build} queued") @@ -90,15 +94,23 @@ class JenkinsClient: table.add_column("Property", style="cyan") table.add_column("Value") - table.add_row("Status", f"[{status_color}]{info['result'] or 'RUNNING'}[/{status_color}]") + table.add_row( + "Status", + f"[{status_color}]{info['result'] or 'RUNNING'}[/{status_color}]", + ) table.add_row("Building", str(info["building"])) table.add_row("Duration", f"{info['duration'] // 1000}s") - table.add_row("Timestamp", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(info["timestamp"] / 1000))) + table.add_row( + "Timestamp", + time.strftime( + "%Y-%m-%d %H:%M:%S", time.localtime(info["timestamp"] / 1000) + ), + ) console.print(table) except jenkins.NotFoundException: - console.print(f"[red]Job or build not found[/red]") + console.print("[red]Job or build not found[/red]") except jenkins.JenkinsException as e: console.print(f"[red]Error getting build status: {e}[/red]") @@ -113,28 +125,99 @@ class JenkinsClient: return build_number = last_build["number"] - console.print(f"Watching build #{build_number} of [cyan]{job_name}[/cyan]...\n") + console.print( + f"Watching build #{build_number} of [cyan]{job_name}[/cyan]...\n" + ) while True: info = self.server.get_build_info(job_name, build_number) status = "RUNNING" if info["building"] else info["result"] - status_color = self._get_status_color(info["result"]) if not info["building"] else "yellow" + status_color = ( + self._get_status_color(info["result"]) + if not info["building"] + else "yellow" + ) - console.print(f" [{status_color}]{status}[/{status_color}] - Duration: {info['duration'] // 1000}s", end="\r") + console.print( + f" [{status_color}]{status}[/{status_color}] - Duration: {info['duration'] // 1000}s", + end="\r", + ) if not info["building"]: console.print() # New line - console.print(f"\nBuild completed with status: [{status_color}]{info['result']}[/{status_color}]") + console.print( + f"\nBuild completed with status: [{status_color}]{info['result']}[/{status_color}]" + ) break time.sleep(interval) except jenkins.NotFoundException: - console.print(f"[red]Job not found[/red]") + console.print("[red]Job not found[/red]") except jenkins.JenkinsException as e: console.print(f"[red]Error watching build: {e}[/red]") + def list_running_jobs(self, count_only: bool = False): + """List all currently running jobs.""" + try: + jobs = self.server.get_all_jobs() + running_count = 0 + + for job in jobs: + color = job.get("color", "") + # Check if job is running (color contains "anime") + if "anime" in color: + running_count += 1 + + if count_only: + print(running_count) + return running_count + + if running_count == 0: + console.print("[green]No jobs currently running[/green]") + return 0 + + running_jobs = [] + for job in jobs: + color = job.get("color", "") + if "anime" in color: + job_info = self.server.get_job_info(job["name"]) + last_build = job_info.get("lastBuild") + if last_build: + build_info = self.server.get_build_info( + job["name"], last_build["number"] + ) + running_jobs.append( + { + "name": job["name"], + "build_number": last_build["number"], + "duration": build_info.get("duration", 0) // 1000, + "url": job["url"], + } + ) + + table = Table(title=f"Running Jobs ({len(running_jobs)})") + table.add_column("Name", style="cyan") + table.add_column("Build", style="yellow") + table.add_column("Duration", style="green") + table.add_column("URL") + + for job in running_jobs: + table.add_row( + job["name"], + f"#{job['build_number']}", + f"{job['duration']}s", + job["url"], + ) + + console.print(table) + return len(running_jobs) + + except jenkins.JenkinsException as e: + console.print(f"[red]Error listing running jobs: {e}[/red]") + return -1 + def _get_status_color(self, result: Optional[str]) -> str: """Get color for build status.""" if result is None: