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