Source code for galaxy.tools.deps.requirements

import copy

import six

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


DEFAULT_REQUIREMENT_TYPE = "package"
DEFAULT_REQUIREMENT_VERSION = None


[docs]@six.python_2_unicode_compatible class ToolRequirement(object): """ 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. """
[docs] def __init__(self, name=None, type=None, version=None, specs=[]): self.name = name self.type = type self.version = version self.specs = specs
[docs] 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)
[docs] def copy(self): return copy.deepcopy(self)
[docs] @staticmethod def from_dict(dict): version = dict.get("version", None) name = dict.get("name", None) type = dict.get("type", None) specs = [RequirementSpecification.from_dict(s) for s in dict.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 "ToolRequirement[%s,version=%s,type=%s,specs=%s]" % (self.name, self.version, self.type, self.specs) __repr__ = __str__
[docs]class RequirementSpecification(object): """Refine a requirement using a URI."""
[docs] 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]
[docs] def to_dict(self): return dict(uri=self.uri, version=self.version)
[docs] @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))
[docs]class ToolRequirements(object): """ Represents all requirements (packages, env vars) needed to run a tool. """
[docs] 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()
[docs] @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'])
[docs] def to_list(self): return [r.to_dict() for r in self.tool_requirements]
[docs] 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): for r in self.tool_requirements: yield r 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])
[docs]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.
[docs]@six.python_2_unicode_compatible class ContainerDescription(object):
[docs] def __init__( self, identifier=None, type=DEFAULT_CONTAINER_TYPE, resolve_dependencies=DEFAULT_CONTAINER_RESOLVE_DEPENDENCIES, shell=DEFAULT_CONTAINER_SHELL, ): self.identifier = identifier self.type = type self.resolve_dependencies = resolve_dependencies self.shell = shell
[docs] def to_dict(self): return dict( identifier=self.identifier, type=self.type, resolve_dependencies=self.resolve_dependencies, shell=self.shell, )
[docs] @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 "ContainerDescription[identifier=%s,type=%s]" % (self.identifier, self.type)
[docs]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]
[docs]def parse_requirements_from_xml(xml_root): """ >>> from xml.etree import ElementTree >>> def load_requirements( contents ): ... contents_document = '''<tool><requirements>%s</requirements></tool>''' ... root = ElementTree.fromstring( 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
[docs]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