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 <noreply@anthropic.com>

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()

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: