reloia
f9e33e2847
Build and Push to Registry / build (push) Successful in 29s
- Rename DockerService to DockerContainer and add folder attribute - Move proxy-related modules (models, validator) to services.mitmproxy - Update imports and references to new module locations - Refactor Docker utilities to use DockerContainer and improve service discovery - Add manager.py with create_proxy utility stub - Initialize discovered Docker services at app startup and warn if none found
76 lines
1.9 KiB
Python
76 lines
1.9 KiB
Python
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from integrations.docker.models import DockerContainer
|
|
from integrations.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[DockerContainer] | None:
|
|
services: list[DockerContainer] = []
|
|
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, "folder": folder})
|
|
|
|
return services
|
|
|
|
|
|
def get_all_services(directory: str | Path) -> list[DockerContainer]:
|
|
path = Path(directory)
|
|
if not path.is_dir():
|
|
return []
|
|
|
|
services: list[DockerContainer] = []
|
|
for sub_folder in _get_visible_folders(path):
|
|
sub_services = get_services(sub_folder)
|
|
if sub_services:
|
|
services.extend(sub_services)
|
|
|
|
return services
|