feat: add Docker service management utilities

app/utils/docker/manager.py

@@ -0,0 +1,100 @@
+import socket
+from pathlib import Path
+
+import yaml
+
+from utils.docker.ports import _get_service_port
+from utils.docker.types import DockerService
+
+COMPOSE_NAMES = (
+    "compose.yml",
+    "compose.yaml",
+    "docker-compose.yml",
+    "docker-compose.yaml"
+)
+
+
+def _get_random_port() -> int:
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(('', 0))
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            port = s.getsockname()[1]
+            return port
+    except Exception as e:
+        raise RuntimeError(f"Could not find an available port: {e}")
+
+
+def get_visible_folders(directory: Path) -> list[Path]:
+    return [
+        p for p in Path(directory).iterdir()
+        if p.is_dir() and not p.name.startswith('.')
+    ]
+
+
+def find_docker_compose(path: Path) -> tuple[dict, str] | None:
+    if not path or not path.is_dir():
+        return None
+
+    for name in COMPOSE_NAMES:
+        file = path / name
+        if file.is_file():
+            return yaml.safe_load(file.read_text()), file.name
+
+    return None
+
+
+def get_services(folder: Path) -> list[DockerService] | None:
+    """
+    Returns a list of all public services by automatically finding the docker file of a fodler
+
+    :param folder:
+    :return:
+6
+    Usage::
+        services = get_services("/root/CApp")
+    """
+    try:
+        docker_compose, _ = find_docker_compose(folder)
+    except TypeError:  # nessun docker_compose nella cartella
+        return None
+
+    if not docker_compose or not docker_compose.get('services'):
+        return None
+
+    services: list[DockerService] = []
+
+    for service, config in docker_compose["services"].items():
+        if not config.get("ports"):
+            continue
+        ports = [p for p in (_get_service_port(port) for port in config["ports"]) if p is not None]
+        if not ports:
+            continue
+        services.append({"name": service, "ports": ports})
+
+    return services or None
+
+
+def get_all_services(directory: str) -> list[DockerService] | None:
+    """
+    Returns a list of all public services in all sub folders of a directory
+    :param directory:
+    :return:
+
+    Usage::
+        services = get_all_services("/root/")
+    """
+    path = Path(directory)
+    if not path.is_dir():
+        return None
+
+    sub_folders = get_visible_folders(path)
+    if not sub_folders:
+        return None
+
+    services = []
+    for sub_folder in sub_folders:
+        if s := get_services(sub_folder):
+            services.append(s)
+
+    return services or None

app/utils/docker/ports.py

@@ -0,0 +1,70 @@
+from pathlib import Path
+
+import yaml
+
+from utils.docker.manager import find_docker_compose
+
+
+def _get_service_port(mapping: str) -> int | None:
+    match mapping.split(":"):
+        case [""]:
+            return None
+        case [port] | [port, _]:
+            return int(port)
+        case ["localhost" | "127.0.0.1", _, _]:
+            return None
+        case [_, port, _]:
+            return int(port)
+        case _:
+            return None
+
+
+def _replace_port_in_mapping(mapping: str, port_from: int, port_to: int) -> str:
+    parts = str(mapping).split(":")
+    match parts:
+        case [_]:
+            # "8080" -> "9090"
+            return str(port_to)
+        case [_, container_port]:
+            # "8080:80" -> "9090:80"
+            return f"{port_to}:{container_port}"
+        case [ip, _, container_port]:
+            # "127.0.0.1:8080:80" -> "127.0.0.1:9090:80"
+            return f"{ip}:{port_to}:{container_port}"
+        case _:
+            return str(port_to)
+
+
+def _get_port_index(mappings: list[str], port: int) -> int | None:
+    for i, mapping in enumerate(mappings):
+        if _get_service_port(mapping) == port:
+            return i
+    return None
+
+
+def change_port(
+        folder: Path,
+        service_name: str,
+        port_from: int,
+        port_to: int,
+):
+    docker_compose, filename = find_docker_compose(folder)
+    if not docker_compose or not docker_compose.get('services'):
+        return
+
+    compose_path = folder / filename
+    backup_compose_path = folder / filename.replace(".yml", ".bak.yml")
+
+    compose_path.copy(backup_compose_path)
+
+    for service, config in docker_compose["services"].items():
+        if service != service_name:
+            continue
+        idx = _get_port_index(config["ports"], port_from)
+        if idx is None:
+            return
+        config["ports"][idx] = _replace_port_in_mapping(config["ports"][idx], port_from, port_to)
+
+    compose_path.write_text(yaml.dump(docker_compose, default_flow_style=False))
+
+    # TODO: run docker compose commands

app/utils/docker/types.py

@@ -0,0 +1,6 @@
+from typing import TypedDict
+
+
+class DockerService(TypedDict):
+    name: str
+    ports: list[int]