From aeb4fce709c79036abdfef38b6785fc12eeab453 Mon Sep 17 00:00:00 2001 From: hhf Date: Thu, 5 Dec 2024 22:22:40 +0530 Subject: [PATCH] Add docker_wrapper.py --- docker_wrapper.py | 253 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 docker_wrapper.py diff --git a/docker_wrapper.py b/docker_wrapper.py new file mode 100644 index 0000000..240e90e --- /dev/null +++ b/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}") \ No newline at end of file