Merge pull request 'feature/docker-utils' (#5) from feature/docker-utils into master

Reviewed-on: #5

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/models.py

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

app/utils/docker/ports.py

@@ -0,0 +1,65 @@
+from collections.abc import Mapping
+
+
+def _get_service_port(mapping: str | int | Mapping[str, object]) -> int | None:
+    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, _]:
+            try:
+                return int(port)
+            except ValueError:
+                return None
+        case ["localhost" | "127.0.0.1", _, _]:
+            return None
+        case [_, port, _]:
+            try:
+                return int(port)
+            except ValueError:
+                return None
+        case _:
+            return None
+
+
+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 [_]:
+            return str(port_to)
+        case [_, container_port]:
+            return f"{port_to}:{container_port}"
+        case [ip, _, container_port]:
+            return f"{ip}:{port_to}:{container_port}"
+        case _:
+            return str(port_to)
+
+
+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
+

app/utils/proxy/types.py

@@ -2,5 +2,5 @@ from typing import TypedDict
 
 
 class ServiceConfig(TypedDict):
-    address: str
+    name: str
     port: int

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "motor>=3.7.1",
     "passlib[bcrypt]>=1.7.4",
     "pydantic-settings>=2.13.1",
+    "pyyaml>=6.0.3",
     "redis>=7.4.0",
 ]
 

uv.lock

@@ -855,6 +855,7 @@ dependencies = [
     { name = "motor" },
     { name = "passlib", extra = ["bcrypt"] },
     { name = "pydantic-settings" },
+    { name = "pyyaml" },
     { name = "redis" },
 ]
 
@@ -871,6 +872,7 @@ requires-dist = [
     { name = "motor", specifier = ">=3.7.1" },
     { name = "passlib", extras = ["bcrypt"], specifier = ">=1.7.4" },
     { name = "pydantic-settings", specifier = ">=2.13.1" },
+    { name = "pyyaml", specifier = ">=6.0.3" },
     { name = "redis", specifier = ">=7.4.0" },
 ]