view env/lib/python3.9/site-packages/galaxy/tool_util/deps/requirements.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

import copy


from galaxy.util import (
    asbool,
    xml_text,
)
from galaxy.util.oset import OrderedSet


DEFAULT_REQUIREMENT_TYPE = "package"
DEFAULT_REQUIREMENT_VERSION = None


class ToolRequirement:
    """
    Represents an external requirement that must be available for the tool to
    run (for example, a program, package, or library).  Requirements can
    optionally assert a specific version.
    """

    def __init__(self, name=None, type=None, version=None, specs=None):
        if specs is None:
            specs = []
        self.name = name
        self.type = type
        self.version = version
        self.specs = specs

    def to_dict(self):
        specs = [s.to_dict() for s in self.specs]
        return dict(name=self.name, type=self.type, version=self.version, specs=specs)

    def copy(self):
        return copy.deepcopy(self)

    @staticmethod
    def from_dict(d):
        version = d.get("version")
        name = d.get("name")
        type = d.get("type")
        specs = [RequirementSpecification.from_dict(s) for s in d.get("specs", [])]
        return ToolRequirement(name=name, type=type, version=version, specs=specs)

    def __eq__(self, other):
        return self.name == other.name and self.type == other.type and self.version == other.version and self.specs == other.specs

    def __ne__(self, other):
        return not self.__eq__(other)

    def __hash__(self):
        return hash((self.name, self.type, self.version, frozenset(self.specs)))

    def __str__(self):
        return f"ToolRequirement[{self.name},version={self.version},type={self.type},specs={self.specs}]"

    __repr__ = __str__


class RequirementSpecification:
    """Refine a requirement using a URI."""

    def __init__(self, uri, version=None):
        self.uri = uri
        self.version = version

    @property
    def specifies_version(self):
        return self.version is not None

    @property
    def short_name(self):
        return self.uri.split("/")[-1]

    def to_dict(self):
        return dict(uri=self.uri, version=self.version)

    @staticmethod
    def from_dict(dict):
        uri = dict.get("uri")
        version = dict.get("version", None)
        return RequirementSpecification(uri=uri, version=version)

    def __eq__(self, other):
        return self.uri == other.uri and self.version == other.version

    def __ne__(self, other):
        return not self.__eq__(other)

    def __hash__(self):
        return hash((self.uri, self.version))


class ToolRequirements:
    """
    Represents all requirements (packages, env vars) needed to run a tool.
    """

    def __init__(self, tool_requirements=None):
        if tool_requirements:
            if not isinstance(tool_requirements, list):
                raise ToolRequirementsException('ToolRequirements Constructor expects a list')
            self.tool_requirements = OrderedSet([r if isinstance(r, ToolRequirement) else ToolRequirement.from_dict(r) for r in tool_requirements])
        else:
            self.tool_requirements = OrderedSet()

    @staticmethod
    def from_list(requirements):
        return ToolRequirements(requirements)

    @property
    def resolvable(self):
        return ToolRequirements([r for r in self.tool_requirements if r.type in {'package', 'set_environment'}])

    @property
    def packages(self):
        return ToolRequirements([r for r in self.tool_requirements if r.type == 'package'])

    def to_list(self):
        return [r.to_dict() for r in self.tool_requirements]

    def append(self, requirement):
        if not isinstance(requirement, ToolRequirement):
            requirement = ToolRequirement.from_dict(requirement)
        self.tool_requirements.add(requirement)

    def __eq__(self, other):
        return len(self.tool_requirements & other.tool_requirements) == len(self.tool_requirements) == len(other.tool_requirements)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __iter__(self):
        yield from self.tool_requirements

    def __getitem__(self, ii):
        return list(self.tool_requirements)[ii]

    def __len__(self):
        return len(self.tool_requirements)

    def __hash__(self):
        return sum([r.__hash__() for r in self.tool_requirements])

    def to_dict(self):
        return [r.to_dict() for r in self.tool_requirements]


class ToolRequirementsException(Exception):
    pass


DEFAULT_CONTAINER_TYPE = "docker"
DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES = False
DEFAULT_CONTAINER_SHELL = "/bin/sh"  # Galaxy assumes bash, but containers are usually thinner.


class ContainerDescription:

    def __init__(
        self,
        identifier=None,
        type=DEFAULT_CONTAINER_TYPE,
        resolve_dependencies=DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES,
        shell=DEFAULT_CONTAINER_SHELL,
    ):
        # Force to lowercase because container image names must be lowercase
        self.identifier = identifier.lower() if identifier else None
        self.type = type
        self.resolve_dependencies = resolve_dependencies
        self.shell = shell
        self.explicit = False

    def to_dict(self, *args, **kwds):
        return dict(
            identifier=self.identifier,
            type=self.type,
            resolve_dependencies=self.resolve_dependencies,
            shell=self.shell,
        )

    @staticmethod
    def from_dict(dict):
        identifier = dict["identifier"]
        type = dict.get("type", DEFAULT_CONTAINER_TYPE)
        resolve_dependencies = dict.get("resolve_dependencies", DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES)
        shell = dict.get("shell", DEFAULT_CONTAINER_SHELL)
        return ContainerDescription(
            identifier=identifier,
            type=type,
            resolve_dependencies=resolve_dependencies,
            shell=shell,
        )

    def __str__(self):
        return f"ContainerDescription[identifier={self.identifier},type={self.type}]"


def parse_requirements_from_dict(root_dict):
    requirements = root_dict.get("requirements", [])
    containers = root_dict.get("containers", [])
    return ToolRequirements.from_list(requirements), [ContainerDescription.from_dict(c) for c in containers]


def parse_requirements_from_xml(xml_root):
    """

    >>> from galaxy.util import parse_xml_string
    >>> def load_requirements(contents):
    ...     contents_document = '''<tool><requirements>%s</requirements></tool>'''
    ...     root = parse_xml_string(contents_document % contents)
    ...     return parse_requirements_from_xml(root)
    >>> reqs, containers = load_requirements('''<requirement>bwa</requirement>''')
    >>> reqs[0].name
    'bwa'
    >>> reqs[0].version is None
    True
    >>> reqs[0].type
    'package'
    >>> reqs, containers = load_requirements('''<requirement type="binary" version="1.3.3">cufflinks</requirement>''')
    >>> reqs[0].name
    'cufflinks'
    >>> reqs[0].version
    '1.3.3'
    >>> reqs[0].type
    'binary'
    """
    requirements_elem = xml_root.find("requirements")

    requirement_elems = []
    if requirements_elem is not None:
        requirement_elems = requirements_elem.findall('requirement')

    requirements = ToolRequirements()
    for requirement_elem in requirement_elems:
        name = xml_text(requirement_elem)
        type = requirement_elem.get("type", DEFAULT_REQUIREMENT_TYPE)
        version = requirement_elem.get("version", DEFAULT_REQUIREMENT_VERSION)
        requirement = ToolRequirement(name=name, type=type, version=version)
        requirements.append(requirement)

    container_elems = []
    if requirements_elem is not None:
        container_elems = requirements_elem.findall('container')

    containers = [container_from_element(c) for c in container_elems]

    return requirements, containers


def container_from_element(container_elem):
    identifier = xml_text(container_elem)
    type = container_elem.get("type", DEFAULT_CONTAINER_TYPE)
    resolve_dependencies = asbool(container_elem.get("resolve_dependencies", DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES))
    shell = container_elem.get("shell", DEFAULT_CONTAINER_SHELL)
    container = ContainerDescription(
        identifier=identifier,
        type=type,
        resolve_dependencies=resolve_dependencies,
        shell=shell,
    )
    return container