Add docker_wrapper.py

docker_wrapper.py

@@ -0,0 +1,253 @@
+from subprocess import getoutput, getstatusoutput
+from typing import Dict, List, Tuple, Optional
+import os
+import logging
+from dataclasses import dataclass
+
+
+class DockerError(Exception):
+    """Base exception for Docker-related errors."""
+    pass
+
+
+class NoContainersError(DockerError):
+    """Exception raised when no containers are found."""
+    pass
+
+
+@dataclass
+class ContainerInfo:
+    """Container information structure."""
+    container_id: str
+    image: str
+    command: str
+    created: str
+    status: str
+    ports: str
+    names: str
+
+
+class Docker:
+    """Docker operations wrapper class."""
+    
+    # Constants for docker commands
+    CMD_PS = 'docker ps'
+    CMD_PS_ALL = 'docker ps -a'
+    CMD_CONTAINER_LIST = 'docker container list'
+    
+    # Header fields in docker ps output
+    HEADER_FIELDS = [
+        'CONTAINER ID', 'IMAGE', 'COMMAND', 'CREATED',
+        'STATUS', 'PORTS', 'NAMES'
+    ]
+
+    @staticmethod
+    def _parse_docker_output(raw_output: str, all_containers: bool = True) -> Dict[str, Dict[str, str]]:
+        """
+        Parse raw Docker command output into structured format.
+        
+        Args:
+            raw_output: Raw output from docker command
+            all_containers: Whether to include all containers or just running ones
+            
+        Returns:
+            Dictionary of container information
+            
+        Raises:
+            NoContainersError: If no containers are found
+        """
+        if '\n' not in raw_output:
+            error_msg = (
+                'A Docker container is required to run this program. '
+                'Please create a docker container and try again.'
+            )
+            if not all_containers:
+                error_msg = (
+                    'A running Docker container is required to run this program. '
+                    'Please run a docker container and try again.'
+                )
+            raise NoContainersError(error_msg)
+
+        # Parse header and find column positions
+        header = raw_output[:raw_output.find('\n')+1]
+        header_indices = {
+            field: header.find(field) 
+            for field in Docker.HEADER_FIELDS
+        }
+
+        # Parse container information
+        containers_info = {}
+        raw_info = raw_output[raw_output.find('\n')+1:]
+        
+        for line in raw_info.split('\n'):
+            if not line.strip():
+                continue
+                
+            container_name = line.strip().split()[-1]
+            containers_info[container_name] = {}
+            
+            # Extract each field based on header positions
+            for i, field in enumerate(Docker.HEADER_FIELDS):
+                start_idx = header_indices[field]
+                end_idx = (header_indices[Docker.HEADER_FIELDS[i+1]] 
+                          if i + 1 < len(Docker.HEADER_FIELDS) 
+                          else len(header))
+                containers_info[container_name][field] = line[start_idx:end_idx].strip()
+
+        return containers_info
+
+    @staticmethod
+    def running_containers_info() -> Dict[str, Dict[str, str]]:
+        """
+        Get information about all running Docker containers.
+        
+        Returns:
+            Dictionary of running container information
+            
+        Raises:
+            NoContainersError: If no running containers are found
+        """
+        return Docker._parse_docker_output(
+            getoutput(Docker.CMD_PS), 
+            all_containers=False
+        )
+
+    @staticmethod
+    def all_containers_info() -> Dict[str, Dict[str, str]]:
+        """
+        Get information about all Docker containers (running and stopped).
+        
+        Returns:
+            Dictionary of all container information
+            
+        Raises:
+            NoContainersError: If no containers are found
+        """
+        return Docker._parse_docker_output(
+            getoutput(Docker.CMD_PS_ALL), 
+            all_containers=True
+        )
+
+    @staticmethod
+    def containers() -> List[str]:
+        """
+        Get list of all container names.
+        
+        Returns:
+            List of container names
+            
+        Raises:
+            NoContainersError: If no containers are found
+        """
+        raw_info = getoutput(Docker.CMD_CONTAINER_LIST)
+        if '\n' not in raw_info:
+            raise NoContainersError(
+                'A Docker container is required to run this program. '
+                'Please create a docker container and try again.'
+            )
+        
+        return [
+            line.strip().split()[-1] 
+            for line in raw_info.split('\n')[1:] 
+            if line.strip()
+        ]
+
+    @staticmethod
+    def container_info(container: str) -> Dict[str, str]:
+        """
+        Get information about a specific container.
+        
+        Args:
+            container: Name of the container
+            
+        Returns:
+            Dictionary containing container information
+            
+        Raises:
+            NoContainersError: If no containers exist
+            KeyError: If the specified container is not found
+        """
+        info = Docker.all_containers_info()
+        if container not in info:
+            raise KeyError(f"Container '{container}' not found")
+        return info[container]
+
+    @staticmethod
+    def compose(directory: str) -> Tuple[int, str]:
+        """
+        Run docker compose in specified directory.
+        
+        Args:
+            directory: Path to directory containing docker-compose.yml
+            
+        Returns:
+            Tuple of (exit_code, output)
+        """
+        original_dir = os.getcwd()
+        try:
+            os.chdir(directory)
+            return getstatusoutput(
+                'docker compose up --detach --build --remove-orphans'
+            )
+        finally:
+            os.chdir(original_dir)
+
+    @staticmethod
+    def start(container: str) -> int:
+        """
+        Start a Docker container.
+        
+        Args:
+            container: Name of container to start
+            
+        Returns:
+            Exit code from docker start command
+        """
+        return getstatusoutput(f'docker start {container}')[0]
+
+    @staticmethod
+    def stop(container: str) -> int:
+        """
+        Stop a Docker container.
+        
+        Args:
+            container: Name of container to stop
+            
+        Returns:
+            Exit code from docker stop command
+        """
+        return getstatusoutput(f'docker stop {container}')[0]
+
+    @staticmethod
+    def rm(container: str) -> int:
+        """
+        Remove a Docker container.
+        
+        Args:
+            container: Name of container to remove
+            
+        Returns:
+            Exit code from docker rm command
+        """
+        return getstatusoutput(f'docker rm {container}')[0]
+
+    @staticmethod
+    def containers_exist() -> bool:
+        """
+        Check if any Docker containers exist.
+        
+        Returns:
+            True if containers exist, False otherwise
+        """
+        return '\n' in getoutput(Docker.CMD_PS_ALL)
+
+
+if __name__ == '__main__':
+    import pprint
+    
+    try:
+        pprint.pprint(Docker.all_containers_info())
+    except NoContainersError as e:
+        print(f"Error: {e}")
+    except Exception as e:
+        print(f"Unexpected error: {e}")