Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.deps.resolvers.galaxy_packages

import logging
from os import listdir
from os.path import (
    basename,
    exists,
    isdir,
    islink,
    join,
    realpath,
)

from . import (
    Dependency,
    DependencyResolver,
    ListableDependencyResolver,
    MappableDependencyResolver,
    NullDependency,
)
from .resolver_mixins import UsesToolDependencyDirMixin

log = logging.getLogger(__name__)


[docs]class GalaxyPackageDependency(Dependency): dict_collection_visible_keys = Dependency.dict_collection_visible_keys + ['script', 'path', 'version', 'name', 'dependency_resolver'] dependency_type = 'galaxy_package'
[docs] def __init__(self, script, path, name, type, version, exact=True, dependency_resolver=None): self.script = script self.path = path self.name = name self.type = type self.version = version self._exact = exact assert self.script is not None or self.path is not None self.dependency_resolver = dependency_resolver
@property def exact(self): return self._exact
[docs] def shell_commands(self): base_path = self.path if self.type == 'package' and self.script is None: commands = 'PACKAGE_BASE=%s; export PACKAGE_BASE; PATH="%s/bin:$PATH"; export PATH' % (base_path, base_path) else: commands = 'PACKAGE_BASE=%s; export PACKAGE_BASE; . %s' % (base_path, self.script) return commands
[docs]class ToolShedDependency(GalaxyPackageDependency): dependency_type = 'tool_shed_package'
class BaseGalaxyPackageDependencyResolver(DependencyResolver, UsesToolDependencyDirMixin): dict_collection_visible_keys = DependencyResolver.dict_collection_visible_keys + ['base_path', 'versionless'] dependency_type = GalaxyPackageDependency def __init__(self, dependency_manager, **kwds): # Galaxy tool shed requires explicit versions on XML elements, # this in inconvient for testing or Galaxy instances not utilizing # the tool shed so allow a fallback version of the Galaxy package # resolver that will just grab 'default' version of exact version # unavailable. self.versionless = str(kwds.get('versionless', "false")).lower() == "true" self._init_base_path(dependency_manager, **kwds) def resolve(self, requirement, **kwds): """ Attempt to find a dependency named `name` at version `version`. If version is None, return the "default" version as determined using a symbolic link (if found). Returns a triple of: env_script, base_path, real_version """ name, version, type = requirement.name, requirement.version, requirement.type if version is None or self.versionless: exact = not self.versionless or version is None return self._find_dep_default(name, type=type, exact=exact, **kwds) else: return self._find_dep_versioned(name, version, type=type, **kwds) def _find_dep_versioned(self, name, version, type='package', **kwds): base_path = self.base_path path = join(base_path, name, version) return self._galaxy_package_dep(path, version, name, type, True) def _find_dep_default(self, name, type='package', exact=True, **kwds): base_path = self.base_path path = join(base_path, name, 'default') if islink(path): real_path = realpath(path) real_version = basename(real_path) return self._galaxy_package_dep(real_path, real_version, name, type, exact) else: return NullDependency(version=None, name=name) def _galaxy_package_dep(self, path, version, name, type, exact): script = join(path, 'env.sh') if exists(script): return self.dependency_type(script, path, name, type, version, exact, dependency_resolver=self) elif exists(join(path, 'bin')): return self.dependency_type(None, path, name, type, version, exact, dependency_resolver=self) return NullDependency(version=version, name=name)
[docs]class GalaxyPackageDependencyResolver(BaseGalaxyPackageDependencyResolver, ListableDependencyResolver, MappableDependencyResolver): resolver_type = "galaxy_packages"
[docs] def __init__(self, dependency_manager, **kwds): super(GalaxyPackageDependencyResolver, self).__init__(dependency_manager, **kwds) self._setup_mapping(dependency_manager, **kwds)
[docs] def resolve(self, requirement, **kwds): requirement = self._expand_mappings(requirement) return super(GalaxyPackageDependencyResolver, self).resolve(requirement, **kwds)
[docs] def list_dependencies(self): base_path = self.base_path for package_name in listdir(base_path): package_dir = join(base_path, package_name) if isdir(package_dir): for version in listdir(package_dir): version_dir = join(package_dir, version) if version == "default": version = None valid_dependency = _is_dependency_directory(version_dir) if valid_dependency: yield self._to_requirement(package_name, version)
def _is_dependency_directory(directory): return exists(join(directory, 'env.sh')) or exists(join(directory, 'bin')) __all__ = ( 'GalaxyPackageDependency', 'GalaxyPackageDependencyResolver', 'ToolShedDependency' )