128 lines
No EOL
4.3 KiB
Python
128 lines
No EOL
4.3 KiB
Python
import os
|
|
from pathlib import Path
|
|
from typing import List, Dict
|
|
import logging
|
|
from docker_wrapper import Docker
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""Custom exception for configuration related errors."""
|
|
pass
|
|
|
|
|
|
class DockerComposer:
|
|
def __init__(self, config_file: str = 'docker-composer.conf'):
|
|
"""Initialize DockerComposer with configuration file."""
|
|
self.logger = self._setup_logger()
|
|
self.working_dir = Path.cwd()
|
|
self.config = self._load_config(config_file)
|
|
|
|
def _setup_logger(self) -> logging.Logger:
|
|
"""Set up logging configuration."""
|
|
logger = logging.getLogger('DockerComposer')
|
|
logger.setLevel(logging.INFO)
|
|
|
|
formatter = logging.Formatter('%(message)s')
|
|
handler = logging.StreamHandler()
|
|
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")
|
|
|
|
try:
|
|
with open(config_file, 'rt') as f:
|
|
lines = f.readlines()
|
|
|
|
config = {}
|
|
|
|
# Parse compose path
|
|
compose_path = self._parse_config_line(lines[0], 'compose-path')
|
|
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')
|
|
config['exclude_containers'] = [
|
|
container.strip()
|
|
for container in exclude_str.split(',')
|
|
if container.strip()
|
|
]
|
|
|
|
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_directories(self) -> Dict[str, Path]:
|
|
"""Get valid Docker compose directories."""
|
|
compose_dirs = {}
|
|
compose_path = self.config['compose_path']
|
|
|
|
if not compose_path.exists():
|
|
raise ConfigError(f"Compose path '{compose_path}' does not exist")
|
|
|
|
for item in compose_path.iterdir():
|
|
# Skip if not a directory or hidden
|
|
if not item.is_dir() or item.name.startswith('.'):
|
|
continue
|
|
|
|
# Skip excluded containers
|
|
if item.name in self.config['exclude_containers']:
|
|
self.logger.info(f"Skipping excluded container: {item.name}")
|
|
continue
|
|
|
|
# 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():
|
|
try:
|
|
self.logger.info(f"COMPOSING {container_name}...")
|
|
Docker.compose(str(directory))
|
|
self.logger.info("DONE!")
|
|
except Exception as e:
|
|
self.logger.error(f"Failed to compose {container_name}: {str(e)}")
|
|
|
|
|
|
def main():
|
|
"""Main entry point for the Docker composer."""
|
|
try:
|
|
composer = DockerComposer()
|
|
composer.compose()
|
|
except ConfigError as e:
|
|
print(f"Configuration error: {str(e)}")
|
|
exit(1)
|
|
except Exception as e:
|
|
print(f"Unexpected error: {str(e)}")
|
|
exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |