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 = f'PACKAGE_BASE={base_path}; export PACKAGE_BASE; PATH="{base_path}/bin:$PATH"; export PATH'
        else:
            commands = f"PACKAGE_BASE={base_path}; export PACKAGE_BASE; . {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().__init__(dependency_manager, **kwds)
        self._setup_mapping(dependency_manager, **kwds)

[docs]    def resolve(self, requirement, **kwds):
        requirement = self._expand_mappings(requirement)
        return super().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")