#!/usr/bin/env python3 """ Docker Manager Application This script manages Docker containers defined in subdirectories of `~/docker/`. Each subdirectory is treated as a project and must contain a `docker-compose.yml`, `docker-compose.yaml`, `compose.yml`, or `compose.yaml`. Usage Examples: 1. List all running containers: $ dm list 2. Update a specific project (pulls latest images and recreates containers): $ dm update media-server 3. Stop all projects (with confirmation prompt): $ dm stop 4. Restart all projects without confirmation: $ dm restart --all 5. Stop a specific project: $ dm stop web-app Purpose: - Simplifies management of multiple Docker Compose projects. - Provides a unified interface for common operations (list, stop, update, restart). - Eliminates the need to navigate to specific directories to run docker commands. To make the command available as dm from any directory, run: chmod +x ~/shell/docker-manager/docker-manager.py mkdir -p ~/.local/bin ln -s ~/shell/docker-manager/docker-manager.py ~/.local/bin/dm """ import os import argparse import subprocess import json from pathlib import Path try: from rich.console import Console from rich.table import Table from rich.panel import Panel from rich import box except ImportError: print("Error: The 'rich' library is required but not installed.") print("Please install it using: pip install -r ~/shell/docker-manager/requirements.txt") print("Or directly: pip install rich") exit(1) # Configuration DOCKER_ROOT = Path(os.path.expanduser("~/docker")) console = Console() def get_diun_container_name(): """Find the running diun container name.""" # Try to find by image name first (most reliable if standard image is used) cmd = ["docker", "ps", "--filter", "ancestor=crazymax/diun", "--format", "{{.Names}}"] res = subprocess.run(cmd, capture_output=True, text=True) if res.returncode == 0 and res.stdout.strip(): return res.stdout.strip().split('\n')[0] # Fallback to finding by name containing 'diun' cmd = ["docker", "ps", "--filter", "name=diun", "--format", "{{.Names}}"] res = subprocess.run(cmd, capture_output=True, text=True) if res.returncode == 0 and res.stdout.strip(): return res.stdout.strip().split('\n')[0] return None def get_diun_info(): """Fetch image information from Diun.""" diun_info = {} try: diun_container = get_diun_container_name() if not diun_container: return diun_info # Get diun image list cmd = ["docker", "exec", diun_container, "diun", "image", "list", "--raw"] res = subprocess.run(cmd, capture_output=True, text=True) if res.returncode != 0: return diun_info data = json.loads(res.stdout) for img in data.get("images", []): name = img.get("name", "") latest = img.get("latest", {}) digest = latest.get("digest", "") labels = latest.get("labels", {}) # Normalize name for matching # Remove docker.io/ prefix norm_name = name if norm_name.startswith("docker.io/"): norm_name = norm_name[10:] # If no registry (no /), prepend library/ if "/" not in norm_name: norm_name = f"library/{norm_name}" tag = latest.get("tag", "latest") if norm_name not in diun_info: diun_info[norm_name] = {} diun_info[norm_name][tag] = { "digest": digest, "labels": labels, "original_name": name } except (subprocess.CalledProcessError, json.JSONDecodeError): pass return diun_info def get_image_version(labels): """Extract version from image labels.""" version_keys = [ "org.opencontainers.image.version", "version", "org.label-schema.version", "build_version" ] for key in version_keys: if key in labels: return labels[key] return "latest" def get_projects(): """Scan DOCKER_ROOT for subdirectories with compose files.""" projects = {} if not DOCKER_ROOT.exists(): print(f"Error: Directory {DOCKER_ROOT} does not exist.") return projects for item in DOCKER_ROOT.iterdir(): if item.is_dir(): if (item / "docker-compose.yml").exists() or \ (item / "docker-compose.yaml").exists() or \ (item / "compose.yml").exists() or \ (item / "compose.yaml").exists(): projects[item.name] = item return projects def run_command(cmd, cwd, capture_output=False): """Run a shell command in a specific directory.""" try: result = subprocess.run( cmd, cwd=cwd, check=True, text=True, capture_output=capture_output ) return result except subprocess.CalledProcessError as e: if capture_output: return e print(f"Error running command: {' '.join(cmd)}") print(f"cwd: {cwd}") return None def list_containers(projects): """List running containers for all projects.""" table = Table(title="Docker Containers", box=box.ROUNDED) table.add_column("Project", style="cyan", no_wrap=True) table.add_column("Container Name", style="green") table.add_column("State", style="yellow") table.add_column("Image", style="blue") table.add_column("Version", style="magenta") table.add_column("Update", style="red") # Get Diun info once diun_info = get_diun_info() found_any = False for name, path in sorted(projects.items()): # Get running container names cmd = ["docker", "compose", "ps", "--format", "json"] res = run_command(cmd, path, capture_output=True) if res and res.returncode == 0: lines = res.stdout.strip().split('\n') for line in lines: if not line: continue try: container = json.loads(line) c_name = container.get('Name', '') state = container.get('State', '') image = container.get('Image', '') if not c_name: continue found_any = True # Get version and update status version = "unknown" update_status = "" # Normalize image name for lookup norm_image = image tag = "latest" # Remove tag if present (heuristic: split on last colon, if right side has no slashes) if ":" in norm_image: base, sep, t = norm_image.rpartition(":") if sep and "/" not in t: norm_image = base tag = t if norm_image.startswith("docker.io/"): norm_image = norm_image[10:] if "/" not in norm_image: norm_image = f"library/{norm_image}" # Check Diun info if diun_info: if norm_image in diun_info: image_tags = diun_info[norm_image] # Only proceed if we have info for this specific tag if tag in image_tags: info = image_tags[tag] # Try to get version from Diun labels first version = get_image_version(info.get("labels", {})) # Check for updates # First get the Image ID of the running container inspect_id_cmd = ["docker", "inspect", c_name, "--format", "{{.Image}}"] inspect_id_res = run_command(inspect_id_cmd, path, capture_output=True) if inspect_id_res and inspect_id_res.returncode == 0: image_id = inspect_id_res.stdout.strip() # Now inspect the Image ID to get RepoDigests inspect_digest_cmd = ["docker", "inspect", image_id, "--format", "{{if .RepoDigests}}{{index .RepoDigests 0}}{{end}}"] inspect_digest_res = run_command(inspect_digest_cmd, path, capture_output=True) if inspect_digest_res and inspect_digest_res.returncode == 0: running_digest_full = inspect_digest_res.stdout.strip() # running_digest is like name@sha256:hash if "@" in running_digest_full: running_digest = running_digest_full.split("@")[1] latest_digest = info.get("digest", "") if latest_digest: if running_digest != latest_digest: update_status = "[bold red]Update Available[/bold red]" else: update_status = "[green]Up to Date[/green]" else: update_status = "[dim]Tag Not Monitored[/dim]" else: update_status = "[dim]Not Monitored[/dim]" # If version is still unknown, try to get from running container labels if version in ("latest", "unknown"): inspect_cmd = ["docker", "inspect", c_name, "--format", "{{json .Config.Labels}}"] inspect_res = run_command(inspect_cmd, path, capture_output=True) if inspect_res and inspect_res.returncode == 0: try: labels = json.loads(inspect_res.stdout) version = get_image_version(labels) except Exception: pass table.add_row(name, c_name, state, image, version, update_status) except json.JSONDecodeError: pass if found_any: console.print(table) else: console.print("[yellow]No running containers found in managed projects.[/yellow]") def is_project_running(path): """Check if any containers in the project are running.""" cmd = ["docker", "compose", "ps", "--format", "json", "--filter", "status=running"] res = run_command(cmd, path, capture_output=True) if res and res.returncode == 0: # If output is not empty/just brackets, something is running output = res.stdout.strip() return output and output != "[]" return False def describe_project(projects, target): """Show detailed information about containers in a project.""" if target not in projects: console.print(f"[red]Error: Project '{target}' not found.[/red]") return path = projects[target] console.print(Panel(f"[bold blue]Describing project: {target}[/bold blue]", expand=False)) cmd = ["docker", "compose", "ps", "--format", "json"] res = run_command(cmd, path, capture_output=True) if res and res.returncode == 0: lines = res.stdout.strip().split('\n') found_any = False for line in lines: if not line: continue try: container = json.loads(line) found_any = True name = container.get('Name', 'N/A') # Create a table for this container table = Table(show_header=False, box=box.SIMPLE) table.add_column("Field", style="bold cyan") table.add_column("Value") table.add_row("Service", container.get('Service', 'N/A')) table.add_row("Container", name) table.add_row("Image", container.get('Image', 'N/A')) table.add_row("State", container.get('State', 'N/A')) table.add_row("Status", container.get('Status', 'N/A')) ports = container.get('Ports', '') if ports: table.add_row("Ports", ports) # Get details via inspect inspect_cmd = ["docker", "inspect", name, "--format", "{{json .}}"] inspect_res = run_command(inspect_cmd, path, capture_output=True) if inspect_res and inspect_res.returncode == 0: try: details = json.loads(inspect_res.stdout.strip()) # Description labels = details.get("Config", {}).get("Labels", {}) description = labels.get("org.opencontainers.image.description") if description: table.add_row("Description", description) console.print(Panel(table, title=f"[bold]{name}[/bold]", border_style="green")) # Volumes/Mounts mounts = details.get("Mounts", []) if mounts: v_table = Table(title="Volumes", box=box.MINIMAL) v_table.add_column("Type", style="cyan") v_table.add_column("Source", style="yellow") v_table.add_column("Destination", style="green") for mount in mounts: m_type = mount.get("Type") source = mount.get("Source", "") dest = mount.get("Destination", "") v_table.add_row(m_type, source, dest) console.print(v_table) console.print("") # Spacing except json.JSONDecodeError: pass else: console.print(Panel(table, title=f"[bold]{name}[/bold]", border_style="green")) except json.JSONDecodeError: pass if not found_any: console.print("[yellow]No containers found for this project.[/yellow]") else: console.print("[red]Failed to get container info.[/red]") def list_project_volumes(projects, target): """List volumes used by containers in a project.""" if target not in projects: console.print(f"[red]Error: Project '{target}' not found.[/red]") return path = projects[target] table = Table(title=f"Volumes for project: {target}", box=box.ROUNDED) table.add_column("Service", style="cyan") table.add_column("Container", style="blue") table.add_column("Type", style="magenta") table.add_column("Source", style="yellow") table.add_column("Destination", style="green") cmd = ["docker", "compose", "ps", "--format", "json"] res = run_command(cmd, path, capture_output=True) if res and res.returncode == 0: lines = res.stdout.strip().split('\n') for line in lines: if not line: continue try: container = json.loads(line) name = container.get('Name', 'N/A') service = container.get('Service', 'N/A') inspect_cmd = ["docker", "inspect", name, "--format", "{{json .Mounts}}"] inspect_res = run_command(inspect_cmd, path, capture_output=True) if inspect_res and inspect_res.returncode == 0: try: mounts = json.loads(inspect_res.stdout.strip()) if mounts: for mount in mounts: m_type = mount.get("Type", "N/A") source = mount.get("Source", "N/A") dest = mount.get("Destination", "N/A") table.add_row(service, name, m_type, source, dest) except json.JSONDecodeError: pass except json.JSONDecodeError: pass console.print(table) def manage_project(projects, action, target, extra_args=None, container_filter=None): """ Execute the specified action (stop, update, restart, logs) on target project(s). Args: projects (dict): Dictionary of project names to their paths. action (str): The action to perform ('stop', 'update', 'restart', 'logs'). target (str): The target project name or 'all'. extra_args (list): Additional arguments for the command (e.g. for logs). container_filter (str): Optional container/service name to filter logs. """ targets = [] # Determine which projects to target if target == "all": targets = sorted(projects.items()) elif target in projects: targets = [(target, projects[target])] else: console.print(f"[red]Error: Project '{target}' not found in {DOCKER_ROOT}[/red]") console.print(f"Available projects: {', '.join(sorted(projects.keys()))}") return # Logs are special, usually run on one project interactively if action == "logs": if len(targets) > 1: console.print("[red]Error: Logs can only be viewed for one project at a time.[/red]") return name, path = targets[0] # Filter by container/service if specified services_to_log = [] if container_filter: cmd_ps = ["docker", "compose", "ps", "--format", "json"] res = run_command(cmd_ps, path, capture_output=True) if res and res.returncode == 0: try: lines = res.stdout.strip().split('\n') for line in lines: if not line: continue c = json.loads(line) s_name = c.get('Service', '') c_name = c.get('Name', '') # Check if filter matches service or container name if container_filter in s_name or container_filter in c_name: services_to_log.append(s_name) except json.JSONDecodeError: pass if not services_to_log: console.print(f"[yellow]No matching service/container found for '{container_filter}'. Showing all logs.[/yellow]") else: services_to_log = list(set(services_to_log)) console.print(f"[blue]Showing logs for service(s): {', '.join(services_to_log)}[/blue]") console.print(f"[blue]Viewing logs for {name}...[/blue]") cmd = ["docker", "compose", "logs"] + (extra_args or []) if services_to_log: cmd.extend(services_to_log) # For logs, we want to stream output directly to stdout, so no capture_output # and we might want to allow Ctrl+C to exit gracefully try: subprocess.run(cmd, cwd=path, check=False) except KeyboardInterrupt: console.print("\n[yellow]Log view stopped.[/yellow]") return console.print(f"[bold]Performing '{action}' on {len(targets)} project(s)...[/bold]") for name, path in targets: console.print(f"\n[bold cyan][{name}] -> {action}...[/bold cyan]") # Execute the requested action if action == "stop": # Stop containers without removing them run_command(["docker", "compose", "stop"], path) elif action == "restart": # Restart containers run_command(["docker", "compose", "restart"], path) elif action == "update": # Update process: Check running -> Stop -> Pull -> Start if was running was_running = is_project_running(path) if was_running: console.print(f" [yellow]Stopping running containers for {name}...[/yellow]") run_command(["docker", "compose", "stop"], path) console.print(f" [blue]Pulling latest images for {name}...[/blue]") pull_res = run_command(["docker", "compose", "pull"], path) if pull_res and pull_res.returncode == 0: if was_running: console.print(f" [green]Restarting containers for {name}...[/green]") run_command(["docker", "compose", "up", "-d"], path) else: console.print( f" [green]Project {name} was not running. Images updated, but not started.[/green]") else: console.print(f" [red]Failed to pull images for {name}. Skipping update.[/red]") def main(): """Main entry point for the application.""" parser = argparse.ArgumentParser( description="Manage Docker containers in ~/docker/") subparsers = parser.add_subparsers( dest="command", help="Command to execute") # List command configuration subparsers.add_parser("list", help="List running containers") # Describe command configuration describe_parser = subparsers.add_parser("describe", help="Show details of a project's containers") describe_parser.add_argument("project", help="Project name to describe") # Volumes command configuration volumes_parser = subparsers.add_parser("volumes", help="List volumes used by a project") volumes_parser.add_argument("project", help="Project name to list volumes for") # Stop command configuration stop_parser = subparsers.add_parser("stop", help="Stop containers") stop_parser.add_argument("project", nargs="?", default=None, help="Project name (optional). If omitted, asks for confirmation to stop ALL.") stop_parser.add_argument("--all", action="store_true", help="Stop all projects without confirmation prompt if specified") # Update command configuration update_parser = subparsers.add_parser( "update", help="Pull and update containers") update_parser.add_argument("project", nargs="?", default=None, help="Project name (optional). If omitted, asks for confirmation to update ALL.") update_parser.add_argument("--all", action="store_true", help="Update all projects without confirmation prompt if specified") # Restart command configuration restart_parser = subparsers.add_parser( "restart", help="Restart containers") restart_parser.add_argument("project", nargs="?", default=None, help="Project name (optional). If omitted, asks for confirmation to restart ALL.") restart_parser.add_argument("--all", action="store_true", help="Restart all projects without confirmation prompt if specified") # Logs command configuration logs_parser = subparsers.add_parser("logs", help="View container logs") logs_parser.add_argument("project", help="Project name to view logs for") logs_parser.add_argument("container", nargs="?", help="Container/Service name to filter logs (optional)") logs_parser.add_argument( "-f", "--follow", action="store_true", help="Follow log output") logs_parser.add_argument( "--tail", default="all", help="Number of lines to show from the end of the logs (default: all)") args = parser.parse_args() # Scan for projects projects = get_projects() if not projects: print(f"No docker projects found in {DOCKER_ROOT}") return # Dispatch commands if args.command == "list": list_containers(projects) elif args.command == "describe": describe_project(projects, args.project) elif args.command == "volumes": list_project_volumes(projects, args.project) elif args.command == "logs": # Prepare extra args for logs extra_args = [] if args.follow: extra_args.append("-f") if args.tail: extra_args.extend(["--tail", args.tail]) manage_project(projects, "logs", args.project, extra_args=extra_args, container_filter=args.container) elif args.command in ["stop", "update", "restart"]: target = args.project # Handle "all" logic safely with user confirmation if target is None: if args.all: target = "all" else: # Interactive check if user wants to apply to all print( f"No project specified. Do you want to {args.command} ALL projects? (y/N)") choice = input().lower() if choice == 'y': target = "all" else: print("Operation cancelled. Specify a project name or use --all.") return manage_project(projects, args.command, target) else: # Show help if no command provided parser.print_help() if __name__ == "__main__": try: main() finally: console.show_cursor(True)