docker_volumes

gistfile1.txt

#!/usr/bin/env python3
"""
Docker Container and Volume Explorer
Interactive script to list containers and their associated volumes
"""

import subprocess
import json
import sys
from typing import List, Dict, Any

class DockerExplorer:
    def __init__(self):
        self.containers = []
        self.selected_container = None
        
    def run_docker_command(self, command: List[str]) -> Dict[str, Any]:
        """Run a docker command and return parsed JSON output"""
        try:
            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                check=True
            )
            if result.stdout:
                return json.loads(result.stdout)
            return {}
        except subprocess.CalledProcessError as e:
            print(f"Error running command: {' '.join(command)}")
            print(f"Error message: {e.stderr}")
            return {}
        except json.JSONDecodeError:
            return {}
    
    def get_containers(self, include_all: bool = True) -> List[Dict[str, Any]]:
        """Get list of Docker containers"""
        cmd = ["docker", "ps"]
        if include_all:
            cmd.append("-a")
        cmd.extend(["--format", "json"])
        
        containers = []
        result = subprocess.run(cmd, capture_output=True, text=True)
        
        if result.returncode != 0:
            print("Error: Docker command failed. Is Docker running?")
            return []
        
        # Parse each line as separate JSON object
        for line in result.stdout.strip().split('\n'):
            if line:
                try:
                    containers.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
        
        # Get detailed info for each container using inspect
        for container in containers:
            inspect_data = self.run_docker_command([
                "docker", "inspect", container["ID"]
            ])
            if inspect_data:
                container["Details"] = inspect_data[0]
        
        return containers
    
    def display_containers(self, containers: List[Dict[str, Any]]):
        """Display containers in a formatted table"""
        if not containers:
            print("No containers found.")
            return
        
        print("\n" + "=" * 100)
        print(f"{'#':<3} {'CONTAINER NAME':<30} {'STATUS':<20} {'IMAGE':<30}")
        print("=" * 100)
        
        for idx, container in enumerate(containers, 1):
            name = container.get("Names", "N/A")
            status = container.get("Status", "N/A")[:20]
            image = container.get("Image", "N/A")[:29]
            print(f"{idx:<3} {name:<30} {status:<20} {image:<30}")
        
        print("=" * 100)
    
    def select_container(self, containers: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Interactive container selection"""
        while True:
            try:
                choice = input(f"\nSelect container (1-{len(containers)}) or 'q' to quit: ").strip()
                if choice.lower() == 'q':
                    return None
                
                idx = int(choice) - 1
                if 0 <= idx < len(containers):
                    return containers[idx]
                else:
                    print(f"Invalid choice. Please enter a number between 1 and {len(containers)}")
            except ValueError:
                print("Invalid input. Please enter a number or 'q'")
    
    def get_container_volumes(self, container: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Extract volume information from container"""
        mounts = container.get("Details", {}).get("Mounts", [])
        volumes = []
        
        for mount in mounts:
            if mount.get("Type") == "volume":
                volume_info = {
                    "type": "volume",
                    "name": mount.get("Name", "N/A"),
                    "source": mount.get("Source", "N/A"),
                    "destination": mount.get("Destination", "N/A"),
                    "is_anonymous": self.is_anonymous_volume(mount.get("Name", ""))
                }
                volumes.append(volume_info)
            elif mount.get("Type") == "bind":
                volume_info = {
                    "type": "bind",
                    "source": mount.get("Source", "N/A"),
                    "destination": mount.get("Destination", "N/A"),
                    "is_anonymous": False
                }
                volumes.append(volume_info)
        
        return volumes
    
    def is_anonymous_volume(self, volume_name: str) -> bool:
        """Check if a volume is anonymous (hashed ID)"""
        # Anonymous volumes typically have 64-character hex IDs
        return len(volume_name) == 64 and all(c in '0123456789abcdef' for c in volume_name)
    
    def get_volume_details(self, volume_name: str) -> Dict[str, Any]:
        """Get detailed information about a volume"""
        if not volume_name or volume_name == "N/A":
            return {}
        
        return self.run_docker_command([
            "docker", "volume", "inspect", volume_name
        ])
    
    def display_volumes(self, volumes: List[Dict[str, Any]]):
        """Display volumes in a formatted table"""
        if not volumes:
            print("No volumes associated with this container.")
            return
        
        print("\n" + "=" * 100)
        print(f"{'#':<3} {'TYPE':<10} {'NAME/SOURCE':<45} {'DESTINATION':<35}")
        print("=" * 100)
        
        for idx, volume in enumerate(volumes, 1):
            vol_type = volume.get("type", "N/A").upper()
            if volume.get("type") == "volume":
                name = volume.get("name", "N/A")
                # Mark anonymous volumes
                if volume.get("is_anonymous"):
                    name = f"{name} (ANONYMOUS)"
                source_display = name[:44]
            else:
                source_display = volume.get("source", "N/A")[:44]
            
            dest = volume.get("destination", "N/A")[:34]
            
            print(f"{idx:<3} {vol_type:<10} {source_display:<45} {dest:<35}")
        
        print("=" * 100)
    
    def show_volume_actions(self, volumes: List[Dict[str, Any]]):
        """Show available actions for volumes"""
        if not volumes:
            return
        
        print("\n" + "=" * 60)
        print("Available Actions:")
        print("=" * 60)
        print("1. Inspect a specific volume")
        print("2. Show all containers using a specific volume")
        print("3. Show volume mount details")
        print("4. Back to container selection")
        print("5. Exit")
        
        while True:
            choice = input("\nSelect action (1-5): ").strip()
            
            if choice == '1':
                self.inspect_volume(volumes)
            elif choice == '2':
                self.find_containers_using_volume(volumes)
            elif choice == '3':
                self.show_mount_details(volumes)
            elif choice == '4':
                break
            elif choice == '5':
                sys.exit(0)
            else:
                print("Invalid choice. Please enter 1-5")
    
    def inspect_volume(self, volumes: List[Dict[str, Any]]):
        """Inspect a specific volume"""
        # Filter to only named volumes (not bind mounts)
        named_volumes = [v for v in volumes if v.get("type") == "volume" and v.get("name") != "N/A"]
        
        if not named_volumes:
            print("No named volumes found to inspect.")
            return
        
        print("\n" + "=" * 60)
        print("Available Volumes:")
        print("=" * 60)
        for idx, vol in enumerate(named_volumes, 1):
            name = vol.get("name", "N/A")
            if vol.get("is_anonymous"):
                name = f"{name} (ANONYMOUS)"
            print(f"{idx}. {name}")
        
        try:
            choice = int(input("\nSelect volume to inspect (or 0 to cancel): ")) - 1
            if 0 <= choice < len(named_volumes):
                volume_name = named_volumes[choice].get("name")
                details = self.get_volume_details(volume_name)
                if details:
                    print("\n" + "=" * 60)
                    print(f"Volume Details: {volume_name}")
                    print("=" * 60)
                    print(json.dumps(details, indent=2, default=str))
                else:
                    print(f"Could not retrieve details for volume {volume_name}")
        except ValueError:
            print("Invalid input")
    
    def find_containers_using_volume(self, volumes: List[Dict[str, Any]]):
        """Find all containers using a specific volume"""
        named_volumes = [v for v in volumes if v.get("type") == "volume" and v.get("name") != "N/A"]
        
        if not named_volumes:
            print("No named volumes found.")
            return
        
        print("\n" + "=" * 60)
        print("Available Volumes:")
        print("=" * 60)
        for idx, vol in enumerate(named_volumes, 1):
            name = vol.get("name", "N/A")
            if vol.get("is_anonymous"):
                name = f"{name} (ANONYMOUS)"
            print(f"{idx}. {name}")
        
        try:
            choice = int(input("\nSelect volume to search (or 0 to cancel): ")) - 1
            if 0 <= choice < len(named_volumes):
                volume_name = named_volumes[choice].get("name")
                
                # Get all containers using this volume
                result = subprocess.run(
                    ["docker", "ps", "-a", "--filter", f"volume={volume_name}", "--format", "json"],
                    capture_output=True,
                    text=True
                )
                
                containers = []
                for line in result.stdout.strip().split('\n'):
                    if line:
                        try:
                            containers.append(json.loads(line))
                        except json.JSONDecodeError:
                            continue
                
                if containers:
                    print(f"\nContainers using volume '{volume_name}':")
                    print("=" * 60)
                    for container in containers:
                        print(f"• {container.get('Names', 'N/A')} ({container.get('Status', 'N/A')})")
                else:
                    print(f"No containers found using volume '{volume_name}'")
        except ValueError:
            print("Invalid input")
    
    def show_mount_details(self, volumes: List[Dict[str, Any]]):
        """Show detailed mount information"""
        print("\n" + "=" * 60)
        print("Mount Details:")
        print("=" * 60)
        for idx, volume in enumerate(volumes, 1):
            print(f"\n{idx}. {'Volume' if volume.get('type') == 'volume' else 'Bind Mount'}:")
            if volume.get('type') == 'volume':
                print(f"   Name: {volume.get('name')}")
                if volume.get('is_anonymous'):
                    print("   ⚠️  ANONYMOUS VOLUME (auto-generated ID)")
            else:
                print(f"   Source: {volume.get('source')}")
            print(f"   Destination: {volume.get('destination')}")
    
    def run(self):
        """Main interactive loop"""
        print("\n" + "=" * 60)
        print("🐳 DOCKER CONTAINER & VOLUME EXPLORER")
        print("=" * 60)
        
        # Check if Docker is available
        try:
            subprocess.run(["docker", "version"], capture_output=True, check=True)
        except (subprocess.CalledProcessError, FileNotFoundError):
            print("Error: Docker is not available or not running.")
            print("Please ensure Docker is installed and the daemon is running.")
            sys.exit(1)
        
        while True:
            print("\n📦 Fetching containers...")
            self.containers = self.get_containers(include_all=True)
            
            if not self.containers:
                print("No containers found. Create a container first with: docker run hello-world")
                retry = input("\nTry again? (y/n): ").strip().lower()
                if retry != 'y':
                    break
                continue
            
            self.display_containers(self.containers)
            self.selected_container = self.select_container(self.containers)
            
            if not self.selected_container:
                print("\nGoodbye! 👋")
                break
            
            # Get and display volumes
            container_name = self.selected_container.get("Names", "Unknown")
            print(f"\n🔍 Analyzing volumes for container: {container_name}")
            
            volumes = self.get_container_volumes(self.selected_container)
            self.display_volumes(volumes)
            
            # Show available actions
            self.show_volume_actions(volumes)
            
            # Ask if user wants to continue
            continue_choice = input("\n\nSelect another container? (y/n): ").strip().lower()
            if continue_choice != 'y':
                print("\nGoodbye! 👋")
                break

def main():
    explorer = DockerExplorer()
    try:
        explorer.run()
    except KeyboardInterrupt:
        print("\n\n👋 Goodbye!")
        sys.exit(0)
    except Exception as e:
        print(f"\nUnexpected error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()