chore: refactor all docker logic files structure

- Implement compose.py with functions to find compose files, list services, and aggregate service info
- Add editor.py for changing service ports with backup support
- Define DockerService model in models.py
- Refactor ports.py to support new port mapping logic and helpers
- Expose Docker Compose utilities in __init__.py

app/utils/docker/__init__.py

@@ -0,0 +1,17 @@
+from utils.docker.compose import (
+	find_docker_compose,
+	get_all_services,
+	get_services,
+	get_visible_folders,
+)
+from utils.docker.editor import change_port
+from utils.docker.models import DockerService
+
+__all__ = [
+	"DockerService",
+	"find_docker_compose",
+	"get_visible_folders",
+	"get_services",
+	"get_all_services",
+	"change_port",
+]

app/utils/docker/compose.py

@@ -0,0 +1,75 @@
+from pathlib import Path
+
+import yaml
+
+from utils.docker.models import DockerService
+from utils.docker.ports import _get_service_port
+
+COMPOSE_NAMES = (
+    "compose.yml",
+    "compose.yaml",
+    "docker-compose.yml",
+    "docker-compose.yaml",
+)
+
+
+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():
+            content = yaml.safe_load(file.read_text()) or {}
+            return content, file.name
+
+    return None
+
+
+def get_services(folder: Path) -> list[DockerService] | None:
+    services: list[DockerService] = []
+    compose_data = find_docker_compose(folder)
+    if compose_data is None:
+        return None
+
+    docker_compose, _ = compose_data
+    compose_services = docker_compose.get("services") or {}
+    if not compose_services:
+        return None
+
+    for service_name, config in compose_services.items():
+        mappings = config.get("ports") if isinstance(config, dict) else None
+        if not isinstance(mappings, list) or not mappings:
+            continue
+
+        ports = [
+            p
+            for p in (_get_service_port(mapping) for mapping in mappings)
+            if p is not None
+        ]
+        if not ports:
+            continue
+
+        services.append({"name": service_name, "ports": ports})
+
+    return services or None
+
+
+def get_all_services(directory: str | Path) -> list[DockerService] | None:
+    path = Path(directory)
+    if not path.is_dir():
+        return None
+
+    services: list[DockerService] = []
+    for sub_folder in get_visible_folders(path):
+        sub_services = get_services(sub_folder)
+        if sub_services:
+            services.extend(sub_services)
+
+    return services or None

app/utils/docker/editor.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+
+import yaml
+
+from utils.docker.compose import find_docker_compose
+from utils.docker.ports import _get_port_index, _replace_port_in_mapping
+
+
+def _backup_path_for(compose_path: Path) -> Path:
+    if compose_path.suffix in {".yml", ".yaml"}:
+        return compose_path.with_name(f"{compose_path.stem}.bak{compose_path.suffix}")
+    return compose_path.with_name(f"{compose_path.name}.bak")
+
+
+def change_port(
+        folder: Path,
+        service_name: str,
+        port_from: int,
+        port_to: int,
+) -> bool:
+    compose_data = find_docker_compose(folder)
+    if compose_data is None:
+        return False
+
+    docker_compose, filename = compose_data
+    compose_services = docker_compose.get("services") or {}
+    if not compose_services:
+        return False
+
+    service_config = compose_services.get(service_name)
+    if not isinstance(service_config, dict):
+        return False
+
+    mappings = service_config.get("ports")
+    if not isinstance(mappings, list):
+        return False
+
+    idx = _get_port_index(mappings, port_from)
+    if idx is None:
+        return False
+
+    mappings[idx] = _replace_port_in_mapping(mappings[idx], port_from, port_to)
+
+    compose_path = folder / filename
+    backup_path = _backup_path_for(compose_path)
+    backup_path.write_text(compose_path.read_text())
+    compose_path.write_text(yaml.dump(docker_compose, default_flow_style=False, sort_keys=False))
+
+    # TODO: run docker compose commands
+    return True

app/utils/docker/manager.py

@@ -1,100 +0,0 @@
-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

@@ -1,70 +1,65 @@
-from pathlib import Path
+from collections.abc import Mapping
-
-import yaml
-
-from utils.docker.manager import find_docker_compose
 
 
-def _get_service_port(mapping: str) -> int | None:
+def _get_service_port(mapping: str | int | Mapping[str, object]) -> int | None:
-    match mapping.split(":"):
+    if isinstance(mapping, int):
+        return mapping
+
+    if isinstance(mapping, Mapping):
+        published = mapping.get("published")
+        if published is None:
+            return None
+        try:
+            return int(published)
+        except (TypeError, ValueError):
+            return None
+
+    parts = str(mapping).split(":")
+    match parts:
         case [""]:
             return None
         case [port] | [port, _]:
-            return int(port)
+            try:
+                return int(port)
+            except ValueError:
+                return None
         case ["localhost" | "127.0.0.1", _, _]:
             return None
         case [_, port, _]:
-            return int(port)
+            try:
+                return int(port)
+            except ValueError:
+                return None
         case _:
             return None
 
 
-def _replace_port_in_mapping(mapping: str, port_from: int, port_to: int) -> str:
+def _replace_port_in_mapping(
+        mapping: str | int | Mapping[str, object],
+        port_from: int,
+        port_to: int,
+) -> str | dict[str, object]:
+    if isinstance(mapping, Mapping):
+        updated = dict(mapping)
+        if _get_service_port(updated) == port_from:
+            updated["published"] = str(port_to)
+        return updated
+
     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:
+def _get_port_index(mappings: list[str | int | Mapping[str, object]], 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