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