import re from typing import List, Optional from pydantic import ValidationError import compose_viz.spec.compose_spec as spec 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: compose_data: spec.ComposeSpecification try: compose_data = spec.ComposeSpecification.parse_file(file_path) except ValidationError as e: raise RuntimeError(f"Error parsing file '{file_path}': {e}") services: List[Service] = [] assert compose_data.services is not None, "No services found, aborting." for service_name, service_data in compose_data.services.items(): service_name = str(service_name) service_image: Optional[str] = None if service_data.build is not None: if type(service_data.build) is str: service_image = f"build from '{service_data.build}'" elif type(service_data.build) is spec.BuildItem: if service_data.build.context is not None and service_data.build.dockerfile is not None: service_image = ( f"build from '{service_data.build.context}' using '{service_data.build.dockerfile}'" ) elif service_data.build.context is not None: service_image = f"build from '{service_data.build.context}'" if service_data.image is not None: if service_image is not None: service_image += ", image: " + service_data.image else: service_image = service_data.image service_networks: List[str] = [] if service_data.networks is not None: if type(service_data.networks) is spec.ListOfStrings: service_networks = service_data.networks.__root__ elif type(service_data.networks) is dict: service_networks = list(service_data.networks.keys()) service_extends: Optional[Extends] = None if service_data.extends is not None: # https://github.com/compose-spec/compose-spec/blob/master/spec.md#extends # The value of the extends key MUST be a dictionary. assert type(service_data.extends) is spec.Extend service_extends = Extends( service_name=service_data.extends.service, from_file=service_data.extends.file ) service_ports: List[Port] = [] if service_data.ports is not None: for port_data in service_data.ports: host_ip: Optional[str] = None host_port: Optional[str] = None container_port: Optional[str] = None protocol: Optional[str] = None if type(port_data) is float: container_port = str(int(port_data)) host_port = f"0.0.0.0:{container_port}" elif type(port_data) is str: regex = r"(?P\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:)?((?P\d+(\-\d+)?):)?((?P\d+(\-\d+)?))?(/(?P\w+))?" # noqa: E501 match = re.match(regex, port_data) if match: host_ip = match.group("host_ip") host_port = match.group("host_port") container_port = match.group("container_port") protocol = match.group("protocol") assert container_port, "Invalid port format, aborting." if container_port is not None and host_port is None: host_port = container_port if host_ip is not None: host_port = f"{host_ip}{host_port}" else: host_port = f"0.0.0.0:{host_port}" elif type(port_data) is spec.Port: assert port_data.target is not None, "Invalid port format, aborting." # ruamel.yaml does not parse port as int assert type(port_data.published) is not int if type(port_data.published) is str: host_port = port_data.published if type(port_data.target) is int: container_port = str(port_data.target) host_ip = port_data.host_ip protocol = port_data.protocol if container_port is not None and host_port is None: host_port = container_port if host_ip is not None: host_port = f"{host_ip}:{host_port}" else: host_port = f"0.0.0.0:{host_port}" assert host_port is not None, "Error while parsing port, aborting." assert container_port is not None, "Error while parsing port, aborting." if protocol is None: protocol = "any" service_ports.append( Port( host_port=host_port, container_port=container_port, protocol=Protocol[protocol], ) ) service_depends_on: List[str] = [] if service_data.depends_on is not None: if type(service_data.depends_on) is spec.ListOfStrings: service_depends_on = service_data.depends_on.__root__ elif type(service_data.depends_on) is dict: for depends_on in service_data.depends_on.keys(): service_depends_on.append(str(depends_on)) service_volumes: List[Volume] = [] if service_data.volumes is not None: for volume_data in service_data.volumes: if 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], ) ) elif type(volume_data) is spec.ServiceVolume: assert volume_data.target is not None, "Invalid volume input, aborting." # https://github.com/compose-spec/compose-spec/blob/master/spec.md#long-syntax-4 # `volume_data.source` is not applicable for a tmpfs mount. if volume_data.source is None: volume_data.source = volume_data.target assert volume_data.source is not None service_volumes.append( Volume( source=volume_data.source, target=volume_data.target, type=VolumeType[volume_data.type], ) ) service_links: List[str] = [] if service_data.links is not None: service_links = service_data.links cgroup_parent: Optional[str] = None if service_data.cgroup_parent is not None: cgroup_parent = service_data.cgroup_parent container_name: Optional[str] = None if service_data.container_name is not None: container_name = service_data.container_name env_file: List[str] = [] if service_data.env_file is not None: if type(service_data.env_file) is spec.StringOrList: if type(service_data.env_file.__root__) is spec.ListOfStrings: env_file = service_data.env_file.__root__.__root__ elif type(service_data.env_file.__root__) is str: env_file.append(service_data.env_file.__root__) expose: List[str] = [] if service_data.expose is not None: for port in service_data.expose: expose.append(str(port)) profiles: List[str] = [] if service_data.profiles is not None: if type(service_data.profiles) is spec.ListOfStrings: profiles = service_data.profiles.__root__ 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, cgroup_parent=cgroup_parent, container_name=container_name, env_file=env_file, expose=expose, profiles=profiles, ) ) return Compose(services=services)