207 lines
No EOL
7 KiB
Python
207 lines
No EOL
7 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import List, Dict, Optional
|
|
import logging
|
|
from docker_wrapper import Docker, NoContainersError
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""Custom exception for configuration related errors."""
|
|
pass
|
|
|
|
|
|
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.config = self._load_config(config_file)
|
|
self.compose_dirs = self._get_compose_dirs()
|
|
self.containers = self._get_containers()
|
|
|
|
def _setup_logger(self) -> logging.Logger:
|
|
"""Configure logging."""
|
|
logger = logging.getLogger('DockerComposeManager')
|
|
logger.setLevel(logging.DEBUG if self.debug else logging.INFO)
|
|
|
|
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 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 = {}
|
|
working_dir = Path.cwd()
|
|
|
|
# Parse compose path
|
|
compose_path = lines[0][lines[0].find('=')+1:].strip()
|
|
compose_path = Path(compose_path if compose_path.startswith('/')
|
|
else working_dir / compose_path)
|
|
config['compose_path'] = compose_path.resolve()
|
|
|
|
# Parse exclude containers
|
|
exclude_str = lines[1][lines[1].find('=')+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: {working_dir}")
|
|
self.logger.debug(f"Compose path: {config['compose_path']}")
|
|
self.logger.debug(f"Exclude containers: {config['exclude_containers']}")
|
|
|
|
return config
|
|
|
|
except Exception as e:
|
|
raise ConfigError(f"Failed to load configuration: {str(e)}")
|
|
|
|
def _get_compose_dirs(self) -> List[Path]:
|
|
"""Get list of valid compose directories."""
|
|
compose_dirs = []
|
|
|
|
try:
|
|
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']):
|
|
compose_dirs.append(item)
|
|
|
|
if self.debug:
|
|
self.logger.debug(f"Compose directories: {compose_dirs}")
|
|
|
|
return sorted(compose_dirs)
|
|
|
|
except Exception as e:
|
|
raise ConfigError(f"Failed to get compose directories: {str(e)}")
|
|
|
|
def _get_containers(self) -> List[str]:
|
|
"""Get list of container names from compose directories."""
|
|
return [dir.name for dir in self.compose_dirs]
|
|
|
|
def display_menu(self) -> None:
|
|
"""Display container selection menu."""
|
|
print("\nWhat Docker container would you like to (re-)compose?")
|
|
for idx, container in enumerate(self.containers):
|
|
print(f" {idx} - {container}")
|
|
print(" q - quit")
|
|
|
|
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]
|
|
|
|
self.logger.info(f"\nRecomposing {container}...")
|
|
|
|
# Stop and remove existing container if it exists
|
|
if Docker.containers_exist():
|
|
self.logger.info("Stopping container...")
|
|
Docker.stop(container)
|
|
self.logger.info("Removing container...")
|
|
Docker.rm(container)
|
|
|
|
# Compose new container
|
|
self.logger.info("Composing container...")
|
|
status, output = Docker.compose(str(container_dir))
|
|
|
|
if status == 0:
|
|
self.logger.info("Container recomposed successfully!")
|
|
return True
|
|
else:
|
|
self.logger.error(f"Failed to recompose container: {output}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error recomposing container: {str(e)}")
|
|
return False
|
|
|
|
def run(self) -> None:
|
|
"""Main application loop."""
|
|
try:
|
|
while True:
|
|
self.display_menu()
|
|
selection = input().strip()
|
|
|
|
if selection.lower() == 'q':
|
|
self.logger.info("\nExiting...")
|
|
sys.exit(0)
|
|
|
|
try:
|
|
container_index = int(selection)
|
|
if 0 <= container_index < len(self.containers):
|
|
if self.recompose_container(container_index):
|
|
response = input('\nWould you like to (re-)compose another container? (y/N) ').strip().lower()
|
|
if response not in ['y', 'yes']:
|
|
self.logger.info("\nExiting...")
|
|
break
|
|
else:
|
|
self.logger.error("Invalid container index")
|
|
except ValueError:
|
|
self.logger.error("Invalid selection. Please enter a number or 'q'")
|
|
|
|
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."""
|
|
try:
|
|
manager = DockerComposeManager(debug=False)
|
|
manager.run()
|
|
except ConfigError as e:
|
|
print(f"Configuration error: {str(e)}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Unexpected error: {str(e)}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |