maestro/docker_wrapper.py

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}")