diff --git a/app/utils/docker/__init__.py b/app/utils/docker/__init__.py new file mode 100644 index 0000000..524a2c9 --- /dev/null +++ b/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", +] diff --git a/app/utils/docker/compose.py b/app/utils/docker/compose.py new file mode 100644 index 0000000..e6157e5 --- /dev/null +++ b/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 diff --git a/app/utils/docker/editor.py b/app/utils/docker/editor.py new file mode 100644 index 0000000..59ef637 --- /dev/null +++ b/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 diff --git a/app/utils/docker/models.py b/app/utils/docker/models.py new file mode 100644 index 0000000..8644ac5 --- /dev/null +++ b/app/utils/docker/models.py @@ -0,0 +1,6 @@ +from typing import TypedDict + + +class DockerService(TypedDict): + name: str + ports: list[int] diff --git a/app/utils/docker/ports.py b/app/utils/docker/ports.py new file mode 100644 index 0000000..e5cecbc --- /dev/null +++ b/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 + diff --git a/app/utils/proxy/types.py b/app/utils/proxy/types.py index a677c3a..548505d 100644 --- a/app/utils/proxy/types.py +++ b/app/utils/proxy/types.py @@ -2,5 +2,5 @@ from typing import TypedDict class ServiceConfig(TypedDict): - address: str + name: str port: int diff --git a/pyproject.toml b/pyproject.toml index da2043d..9eb71e7 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/uv.lock b/uv.lock index 2f6ed9a..3aa63d3 100644 --- a/uv.lock +++ b/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" }, ]