"""Docker Compose file generation and manipulation.""" import logging import subprocess from collections import Counter from pathlib import Path from typing import Any logger = logging.getLogger(__name__) def sort_volumes_by_specificity(volumes: list[str]) -> list[str]: """Sort volume strings so parent paths come before child paths. Docker Compose mounts volumes in array order. A later mount at a parent path hides earlier mounts at child paths. By sorting shallow paths first and deep paths last, deeper (more specific) mounts overlay correctly. Volume format: source:target or source:target:type Args: volumes: List of Docker volume mount strings. Returns: Sorted list with parent paths before child paths. """ def _target_depth(vol: str) -> int: parts = vol.split(":") if len(parts) < 2: return 0 target = parts[1].rstrip("/") if not target or target == "/": return 0 return target.count("/") # Detect duplicate targets and warn targets = [] for vol in volumes: parts = vol.split(":") targets.append(parts[1] if len(parts) > 1 else "") dupes = [t for t, c in Counter(targets).items() if c > 1] if dupes: logger.warning("Duplicate mount targets detected: %s", dupes) # Stable sort: parent paths first, child paths last return sorted(volumes, key=_target_depth) def render_compose_template(template: str, variables: dict[str, Any]) -> str: """Render a Docker Compose template with variable substitution. Args: template: The compose template string variables: Dictionary of variable names to values Returns: Rendered compose file content """ result = template for key, value in variables.items(): placeholder = f"{{{{{key}}}}}" result = result.replace(placeholder, str(value)) # Convenience aliases so legacy and migrated templates can use lowercase # placeholders without changing every stored template. aliases = { "{{workspace_name}}": "WORKSPACE_NAME", "{{home_directory}}": "HOME_DIRECTORY", } for alias_placeholder, key in aliases.items(): if alias_placeholder in result: result = result.replace( alias_placeholder, str(variables.get(key, "workspace")) ) return result def write_compose_file(instance_dir: str, content: str) -> str: """Write the rendered compose file to the instance directory. Args: instance_dir: Path to instance directory content: Rendered compose content Returns: Path to the compose file """ compose_path = Path(instance_dir) / "docker-compose.yml" compose_path.write_text(content) return str(compose_path) def execute_compose_command( compose_path: str, action: str, timeout: int = 60, env_file: str | None = None ) -> tuple[int, str, str]: """Execute a docker compose command. Args: compose_path: Path to docker-compose.yml action: The compose action (up, down, start, stop, restart) timeout: Command timeout in seconds env_file: Optional path to .env file for environment variables Returns: Tuple of (returncode, stdout, stderr) """ instance_dir = Path(compose_path).parent cmd = ["docker", "compose", "-f", compose_path] if env_file: cmd.extend(["--env-file", env_file]) if action == "up": cmd.extend(["up", "-d", "--force-recreate"]) elif action == "down": cmd.extend(["down", "-v"]) elif action in ("start", "stop", "restart"): cmd.append(action) else: raise ValueError(f"Unknown compose action: {action}") result = subprocess.run( cmd, cwd=str(instance_dir), capture_output=True, text=True, timeout=timeout, ) return result.returncode, result.stdout, result.stderr