compose-viz/compose_viz/parser.py

import re
from typing import Dict, List, Optional

from ruamel.yaml import YAML

from compose_viz.models.compose import Compose, Service
from compose_viz.models.extends import Extends
from compose_viz.models.port import Port, Protocol
from compose_viz.models.volume import Volume, VolumeType


class Parser:
    def __init__(self):
        pass

    def parse(self, file_path: str) -> Compose:
        # load the yaml file
        with open(file_path, "r") as f:
            try:
                yaml = YAML(typ="safe", pure=True)
                yaml_data = yaml.load(f)
            except Exception as e:
                raise RuntimeError(f"Error parsing file '{file_path}': {e}")
        # validate the yaml file
        if not yaml_data:
            raise RuntimeError("Empty yaml file, aborting.")
        if not yaml_data.get("services"):
            raise RuntimeError("No services found, aborting.")

        # parse services data into Service objects
        services = self.parse_service_data(yaml_data["services"])

        # create Compose object
        compose = Compose(services)

        return compose

    def parse_service_data(self, services_yaml_data: Dict[str, dict]) -> List[Service]:
        services: List[Service] = []
        for service, service_name in zip(services_yaml_data.values(), services_yaml_data.keys()):

            service_image: Optional[str] = None
            if service.get("build"):
                if type(service["build"]) is str:
                    service_image = f"build from '{service['build']}'"
                elif type(service["build"]) is dict:
                    if service["build"].get("context") and service["build"].get("dockerfile"):
                        service_image = (
                            f"build from '{service['build']['context']}' using '{service['build']['dockerfile']}'"
                        )
                    elif service["build"].get("context"):
                        service_image = f"build from '{service['build']['context']}'"
            if service.get("image"):
                if service_image:
                    service_image += ", image: " + service["image"]
                else:
                    service_image = service["image"]

            service_networks: List[str] = []
            if service.get("networks"):
                if type(service["networks"]) is list:
                    service_networks = service["networks"]
                elif type(service["networks"]) is dict:
                    service_networks = list(service["networks"].keys())

            service_extends: Optional[Extends] = None
            if service.get("extends"):
                assert type(service["extends"]) is dict, "Invalid extends format, aborting."
                assert service["extends"]["service"], "Missing extends service, aborting."
                extend_service_name = str(service["extends"]["service"])

                extend_from_file: Optional[str] = None
                if service["extends"].get("file"):
                    assert service["extends"]["file"], "Missing extends file, aborting."
                    extend_from_file = str(service["extends"]["file"])

                service_extends = Extends(service_name=extend_service_name, from_file=extend_from_file)

            service_ports: List[Port] = []
            if service.get("ports"):
                assert type(service["ports"]) is list
                for port_data in service["ports"]:
                    if type(port_data) is dict:
                        # define a nested function to limit variable scope
                        def long_syntax():
                            assert type(port_data) is dict
                            assert port_data["target"]

                            container_port: str = str(port_data["target"])
                            host_port: str = ""
                            protocol: Protocol = Protocol.any

                            if port_data.get("published"):
                                host_port = str(port_data["published"])
                            else:
                                host_port = container_port

                            if port_data.get("host_ip"):
                                host_ip = str(port_data["host_ip"])
                                host_port = f"{host_ip}:{host_port}"
                            else:
                                host_port = f"0.0.0.0:{host_port}"

                            if port_data.get("protocol"):
                                protocol = Protocol[str(port_data["protocol"])]

                            assert host_port, "Error parsing port, aborting."

                            service_ports.append(
                                Port(
                                    host_port=host_port,
                                    container_port=container_port,
                                    protocol=protocol,
                                )
                            )

                        long_syntax()
                    elif type(port_data) is str:
                        # ports that needs to parse using regex:
                        #     - "3000"
                        #     - "3000-3005"
                        #     - "8000:8000"
                        #     - "9090-9091:8080-8081"
                        #     - "49100:22"
                        #     - "127.0.0.1:8001:8001"
                        #     - "127.0.0.1:5000-5010:5000-5010"
                        #     - "6060:6060/udp"

                        def short_syntax():
                            assert type(port_data) is str
                            regex = r"(?P<host_ip>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:)?((?P<host_port>\d+(\-\d+)?):)?((?P<container_port>\d+(\-\d+)?))?(/(?P<protocol>\w+))?"  # noqa: E501
                            match = re.match(regex, port_data)
                            if match:
                                host_ip: Optional[str] = match.group("host_ip")
                                host_port: Optional[str] = match.group("host_port")
                                container_port: Optional[str] = match.group("container_port")
                                protocol: Optional[str] = match.group("protocol")

                                assert container_port, "Invalid port format, aborting."

                                if container_port and not host_port:
                                    host_port = container_port

                                if host_ip:
                                    host_port = f"{host_ip}{host_port}"
                                else:
                                    host_port = f"0.0.0.0:{host_port}"

                                assert host_port, "Error while parsing port, aborting."

                                if protocol:
                                    service_ports.append(
                                        Port(
                                            host_port=host_port,
                                            container_port=container_port,
                                            protocol=Protocol[protocol],
                                        )
                                    )
                                else:
                                    service_ports.append(
                                        Port(
                                            host_port=host_port,
                                            container_port=container_port,
                                        )
                                    )

                        short_syntax()

            service_depends_on: List[str] = []
            if service.get("depends_on"):
                if type(service["depends_on"]) is list:
                    for depends_on in service["depends_on"]:
                        service_depends_on.append(str(depends_on))
                elif type(service["depends_on"]) is dict:
                    service_depends_on = list(service["depends_on"].keys())

            service_volumes: List[Volume] = []
            if service.get("volumes"):
                assert type(service["volumes"]) is list
                for volume_data in service["volumes"]:
                    if type(volume_data) is dict:
                        assert volume_data["source"] and volume_data["target"], "Invalid volume input, aborting."

                        volume_source: str = str(volume_data["source"])
                        volume_target: str = str(volume_data["target"])
                        volume_type: VolumeType = VolumeType.volume

                        if volume_data.get("type"):
                            volume_type = VolumeType[str(volume_data["type"])]

                        service_volumes.append(Volume(source=volume_source, target=volume_target, type=volume_type))
                    elif type(volume_data) is str:
                        assert ":" in volume_data, "Invalid volume input, aborting."

                        spilt_data = volume_data.split(":")
                        if len(spilt_data) == 2:
                            service_volumes.append(Volume(source=spilt_data[0], target=spilt_data[1]))
                        elif len(spilt_data) == 3:
                            service_volumes.append(
                                Volume(source=spilt_data[0], target=spilt_data[1], access_mode=spilt_data[2])
                            )

            service_links: List[str] = []
            if service.get("links"):
                service_links = service["links"]

            services.append(
                Service(
                    name=service_name,
                    image=service_image,
                    networks=service_networks,
                    extends=service_extends,
                    ports=service_ports,
                    depends_on=service_depends_on,
                    volumes=service_volumes,
                    links=service_links,
                )
            )
            # Service print debug
            # print("--------------------")
            # print("Service name: {}".format(service_name))
            # print("image: {}".format(service_image))
            # print("networks: {}".format(service_networks))
            # print("image: {}".format(service_image))
            # print("extends: {}".format(service_extends))
            # print("ports: {}".format(service_ports))
            # print("depends: {}".format(service_depends_on))

        return services