#!/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 from rich.console import Console from rich.table import Table from rich.panel import Panel from rich.text import Text from rich import box # 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}" diun_info[norm_name] = { "digest": digest, "labels": labels, "tag": latest.get("tag", ""), "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 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 norm_image in diun_info: info = diun_info[norm_image] # Try to get version from Diun labels first version = get_image_version(info.get("labels", {})) # Check for updates inspect_digest_cmd = ["docker", "inspect", c_name, "--format", "{{index .RepoDigests 0}}"] 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 and running_digest != latest_digest: update_status = "Update Available" # If version is still unknown, try to get from running container labels if version == "latest" or version == "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: 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__": main()