Warning
This document is for an old release 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 = 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
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'
)