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
This commit is contained in:
2026-04-12 19:46:48 +00:00
committed by reloia
Unverified
parent 27ca642338
commit 4fa25355e7
6 changed files with 178 additions and 141 deletions
+36 -41
View File
@@ -1,70 +1,65 @@
from pathlib import Path
import yaml
from utils.docker.manager import find_docker_compose
from collections.abc import Mapping
def _get_service_port(mapping: str) -> int | None:
match mapping.split(":"):
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, _]:
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