Fusion
ad09ffa6ec
- Create ToolConfigsPage with tool type selector, config list, and add/edit form - Support both env and file config types - Add route /tool-configs and navigation item - Update API client with tool config endpoints - Build passes successfully
235 lines
6.5 KiB
Python
235 lines
6.5 KiB
Python
"""Docker service for managing tool instances."""
|
|
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
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))
|
|
return result
|
|
|
|
|
|
def ensure_instance_directory(instance_id: str, base_path: str | None = None) -> str:
|
|
"""Create and return the instance directory path.
|
|
|
|
Args:
|
|
instance_id: Unique instance identifier
|
|
base_path: Base directory for all instances (defaults to Settings.instance_base_path)
|
|
|
|
Returns:
|
|
Absolute path to instance directory
|
|
"""
|
|
if base_path is None:
|
|
from src.config import Settings
|
|
base_path = Settings().instance_base_path
|
|
instance_dir = Path(base_path) / instance_id
|
|
instance_dir.mkdir(parents=True, exist_ok=True)
|
|
return str(instance_dir.absolute())
|
|
|
|
|
|
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 write_env_file(instance_dir: str, env_vars: dict[str, str]) -> str:
|
|
"""Write environment variables to a .env file.
|
|
|
|
Args:
|
|
instance_dir: Path to instance directory
|
|
env_vars: Dictionary of env var names to values
|
|
|
|
Returns:
|
|
Path to the env file
|
|
"""
|
|
env_path = Path(instance_dir) / ".env"
|
|
lines = [f'{key}="{value}"' for key, value in env_vars.items()]
|
|
env_path.write_text("\n".join(lines) + "\n")
|
|
return str(env_path)
|
|
|
|
|
|
def write_config_files(instance_dir: str, files: dict[str, str]) -> None:
|
|
"""Write config files to the instance directory.
|
|
|
|
Args:
|
|
instance_dir: Path to instance directory
|
|
files: Dictionary of file paths (relative to instance dir) to content
|
|
"""
|
|
instance_path = Path(instance_dir)
|
|
for file_path, content in files.items():
|
|
# Ensure the path is within the instance directory (security)
|
|
full_path = instance_path / file_path
|
|
try:
|
|
full_path.resolve().relative_to(instance_path.resolve())
|
|
except ValueError:
|
|
raise ValueError(f"File path '{file_path}' escapes instance directory")
|
|
|
|
full_path.parent.mkdir(parents=True, exist_ok=True)
|
|
full_path.write_text(content)
|
|
|
|
|
|
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"])
|
|
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
|
|
|
|
|
|
def get_container_id(instance_name: str) -> str | None:
|
|
"""Get the container ID for a compose service.
|
|
|
|
Args:
|
|
instance_name: The service name in compose
|
|
|
|
Returns:
|
|
Container ID or None if not found
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "ps", "-q", "--filter", f"name={instance_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return result.stdout.strip().split("\n")[0]
|
|
return None
|
|
|
|
|
|
def get_container_name(instance_name: str) -> str | None:
|
|
"""Get the full container name for a compose service.
|
|
|
|
Args:
|
|
instance_name: The service name in compose
|
|
|
|
Returns:
|
|
Container name or None if not found
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "ps", "--format", "{{.Names}}", "--filter", f"name={instance_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return result.stdout.strip().split("\n")[0]
|
|
return None
|
|
|
|
|
|
def get_container_status(container_id: str) -> str:
|
|
"""Get the status of a Docker container.
|
|
|
|
Args:
|
|
container_id: Docker container ID
|
|
|
|
Returns:
|
|
Container status string (running, exited, etc.)
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "inspect", "-f", "{{.State.Status}}", container_id],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return result.stdout.strip()
|
|
return "unknown"
|
|
|
|
|
|
def get_container_logs(container_id: str, tail: int = 100) -> str:
|
|
"""Get the logs of a Docker container.
|
|
|
|
Args:
|
|
container_id: Docker container ID
|
|
tail: Number of lines to return
|
|
|
|
Returns:
|
|
Container logs
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "logs", "--tail", str(tail), container_id],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return result.stdout
|
|
return f"Failed to get logs: {result.stderr}"
|
|
|
|
|
|
def find_free_port(start: int = 10000, end: int = 20000) -> int:
|
|
"""Find a free TCP port in the given range.
|
|
|
|
Args:
|
|
start: Start of port range
|
|
end: End of port range
|
|
|
|
Returns:
|
|
Free port number
|
|
"""
|
|
import socket
|
|
|
|
for port in range(start, end):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
if s.connect_ex(("localhost", port)) != 0:
|
|
return port
|
|
|
|
raise RuntimeError(f"No free port found in range {start}-{end}")
|