From 686cb719332226cdcd297ea5c93dda0f73404f25 Mon Sep 17 00:00:00 2001 From: hhf Date: Thu, 5 Dec 2024 23:38:50 +0530 Subject: [PATCH] Update composer.py --- composer.py | 219 ++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 152 insertions(+), 67 deletions(-) diff --git a/composer.py b/composer.py index 3645a92..9d99c06 100644 --- a/composer.py +++ b/composer.py @@ -1,8 +1,11 @@ +#!/usr/bin/env python3 + import os from pathlib import Path -from typing import List, Dict +from typing import List, Dict, Optional import logging -from docker_wrapper import Docker +from docker_wrapper import Docker, NoContainersError +import sys class ConfigError(Exception): @@ -10,118 +13,200 @@ class ConfigError(Exception): pass -class DockerComposer: - def __init__(self, config_file: str = 'docker-composer.conf'): - """Initialize DockerComposer with configuration file.""" +class DockerComposeManager: + def __init__(self, config_file: str = 'docker-composer.conf', debug: bool = False): + """ + Initialize Docker Compose Manager. + + Args: + config_file: Path to configuration file + debug: Enable debug logging + """ + self.debug = debug self.logger = self._setup_logger() self.working_dir = Path.cwd() self.config = self._load_config(config_file) - + self.compose_dirs = self._get_compose_dirs() + self.containers = self._get_containers() + def _setup_logger(self) -> logging.Logger: - """Set up logging configuration.""" - logger = logging.getLogger('DockerComposer') - logger.setLevel(logging.INFO) + """Configure logging.""" + logger = logging.getLogger('DockerComposeManager') + logger.setLevel(logging.DEBUG if self.debug else logging.INFO) - formatter = logging.Formatter('%(message)s') handler = logging.StreamHandler() + formatter = logging.Formatter('%(message)s') handler.setFormatter(formatter) logger.addHandler(handler) return logger def _load_config(self, config_file: str) -> Dict[str, any]: - """Load and parse configuration file.""" - if not os.path.exists(config_file): - raise ConfigError(f"Configuration file '{config_file}' not found") - + """ + Load configuration from file. + + Args: + config_file: Path to configuration file + + Returns: + Dictionary containing configuration + + Raises: + ConfigError: If configuration file is invalid or missing + """ try: + if not os.path.exists(config_file): + raise ConfigError(f"Configuration file '{config_file}' not found") + with open(config_file, 'rt') as f: lines = f.readlines() + if len(lines) < 2: + raise ConfigError("Incomplete configuration file") + config = {} - + # Parse compose path - compose_path = self._parse_config_line(lines[0], 'compose-path') + compose_line = next((line for line in lines if 'compose-path' in line), None) + if not compose_line: + raise ConfigError("compose-path not found in configuration") + + compose_path = compose_line.split('=')[1].strip() compose_path = Path(compose_path if compose_path.startswith('/') else self.working_dir / compose_path) config['compose_path'] = compose_path.resolve() # Parse exclude containers - exclude_str = self._parse_config_line(lines[1], 'exclude-containers') + exclude_line = next((line for line in lines if 'exclude-containers' in line), None) + if not exclude_line: + raise ConfigError("exclude-containers not found in configuration") + + exclude_str = exclude_line.split('=')[1].strip() config['exclude_containers'] = [ container.strip() for container in exclude_str.split(',') if container.strip() ] + if self.debug: + self.logger.debug(f"Working directory: {self.working_dir}") + self.logger.debug(f"Compose path: {config['compose_path']}") + self.logger.debug(f"Exclude containers: {config['exclude_containers']}") + return config - except IndexError: - raise ConfigError("Configuration file is incomplete") except Exception as e: raise ConfigError(f"Failed to load configuration: {str(e)}") - def _parse_config_line(self, line: str, key: str) -> str: - """Parse a configuration line to extract its value.""" - if '=' not in line: - raise ConfigError(f"Invalid configuration format for {key}") - return line.split('=', 1)[1].strip() + def _get_compose_dirs(self) -> List[Path]: + """Get list of valid compose directories.""" + compose_dirs = [] + + try: + if not self.config['compose_path'].exists(): + raise ConfigError(f"Compose path '{self.config['compose_path']}' does not exist") + + for item in self.config['compose_path'].iterdir(): + if (item.is_dir() and + not item.name.startswith('.') and + item.name not in self.config['exclude_containers']): + # Verify docker-compose.yml exists + if (item / 'docker-compose.yml').exists(): + compose_dirs.append(item) + else: + self.logger.warning(f"Skipping {item.name}: no docker-compose.yml found") - def _get_compose_directories(self) -> Dict[str, Path]: - """Get valid Docker compose directories.""" - compose_dirs = {} - compose_path = self.config['compose_path'] + if self.debug: + self.logger.debug(f"Compose directories: {compose_dirs}") - if not compose_path.exists(): - raise ConfigError(f"Compose path '{compose_path}' does not exist") + return sorted(compose_dirs) - for item in compose_path.iterdir(): - # Skip if not a directory or hidden - if not item.is_dir() or item.name.startswith('.'): - continue + except Exception as e: + raise ConfigError(f"Failed to get compose directories: {str(e)}") - # Skip excluded containers - if item.name in self.config['exclude_containers']: - self.logger.info(f"Skipping excluded container: {item.name}") - continue + def _get_containers(self) -> List[str]: + """Get list of container names from compose directories.""" + return [dir.name for dir in self.compose_dirs] - # Check for docker-compose.yml - if not (item / 'docker-compose.yml').exists(): - self.logger.warning(f"Skipping {item.name}: no docker-compose.yml found") - continue - - compose_dirs[item.name] = item - - return compose_dirs - - def compose(self) -> None: - """Execute Docker compose for all valid directories.""" - compose_dirs = self._get_compose_directories() - - if not compose_dirs: - self.logger.warning("No valid compose directories found") - return - - for container_name, directory in compose_dirs.items(): + def recompose_all(self) -> None: + """Recompose all containers.""" + for i, container in enumerate(self.containers): try: - self.logger.info(f"COMPOSING {container_name}...") - Docker.compose(str(directory)) - self.logger.info("DONE!") + self.logger.info(f"\nRecomposing {container}...") + if self.recompose_container(i): + self.logger.info(f"{container} recomposed successfully!") + else: + self.logger.error(f"Failed to recompose {container}") except Exception as e: - self.logger.error(f"Failed to compose {container_name}: {str(e)}") + self.logger.error(f"Error recomposing {container}: {str(e)}") + + def recompose_container(self, container_index: int) -> bool: + """ + Recompose a specific container. + + Args: + container_index: Index of container to recompose + + Returns: + bool: True if successful, False otherwise + """ + try: + container = self.containers[container_index] + container_dir = self.compose_dirs[container_index] + + # Stop and remove existing container if it exists + if Docker.containers_exist(): + try: + self.logger.info("Stopping container...") + Docker.stop(container) + self.logger.info("Removing container...") + Docker.rm(container) + except Exception as e: + self.logger.warning(f"Container cleanup failed: {e}") + + # Compose new container + self.logger.info("Composing container...") + status, output = Docker.compose(str(container_dir)) + + if status == 0: + return True + else: + self.logger.error(f"Compose failed: {output}") + return False + + except Exception as e: + self.logger.error(f"Error recomposing container: {str(e)}") + return False + + def run(self) -> None: + """Main execution method - recompose all containers.""" + try: + if not self.compose_dirs: + self.logger.warning("No valid compose directories found") + return + + self.logger.info(f"Found {len(self.containers)} containers to compose") + self.recompose_all() + + except KeyboardInterrupt: + self.logger.info("\nOperation cancelled by user") + sys.exit(0) + except Exception as e: + self.logger.error(f"Unexpected error: {str(e)}") + sys.exit(1) def main(): - """Main entry point for the Docker composer.""" + """Main entry point.""" try: - composer = DockerComposer() - composer.compose() + manager = DockerComposeManager(debug=False) + manager.run() except ConfigError as e: - print(f"Configuration error: {str(e)}") - exit(1) + print(f"Configuration error: {str(e)}", file=sys.stderr) + sys.exit(1) except Exception as e: - print(f"Unexpected error: {str(e)}") - exit(1) + print(f"Unexpected error: {str(e)}", file=sys.stderr) + sys.exit(1) if __name__ == "__main__":