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.mulled.mulled_search
#!/usr/bin/env python
import argparse
import json
import logging
import sys
import tempfile
from .mulled_list import get_singularity_containers
from .util import build_target, v2_image_name
try:
from conda.cli.python_api import run_command
except ImportError:
run_command = None
try:
import requests
except ImportError:
requests = None
try:
from whoosh.fields import Schema
from whoosh.fields import TEXT
from whoosh.fields import STORED
from whoosh.index import create_in
from whoosh.qparser import QueryParser
except ImportError:
Schema = TEXT = STORED = create_in = QueryParser = None
QUAY_API_URL = 'https://quay.io/api/v1/repository'
[docs]class QuaySearch():
"""
Tool to search within a quay organization for a given software name.
"""
[docs] def build_index(self):
"""
Create an index to quickly examine the repositories of a given quay.io organization.
"""
# download all information about the repositories from the
# given organization in self.organization
parameters = {'public': 'true', 'namespace': self.organization}
r = requests.get(QUAY_API_URL, headers={
'Accept-encoding': 'gzip'}, params=parameters, timeout=12)
tmp_dir = tempfile.mkdtemp()
schema = Schema(title=TEXT(stored=True), content=STORED)
self.index = create_in(tmp_dir, schema)
json_decoder = json.JSONDecoder()
decoded_request = json_decoder.decode(r.text)
writer = self.index.writer()
for repository in decoded_request['repositories']:
writer.add_document(
title=repository['name'], content=repository['description'])
writer.commit()
[docs] def search_repository(self, search_string, non_strict):
"""
Search Docker containers on quay.io.
Results are displayed with all available versions,
including the complete image name.
"""
# with statement closes searcher after usage.
with self.index.searcher() as searcher:
query = QueryParser(
"title", self.index.schema).parse(search_string)
results = searcher.search(query)
if non_strict:
# look for spelling errors and use suggestions as a search term too
corrector = searcher.corrector("title")
suggestions = corrector.suggest(search_string, limit=2)
# get all repositories with suggested keywords
for suggestion in suggestions:
search_string = "*%s*" % suggestion
query = QueryParser(
"title", self.index.schema).parse(search_string)
results_tmp = searcher.search(query)
results.extend(results_tmp)
out = list()
for result in results:
title = result['title']
for version in self.get_additional_repository_information(title):
out.append({'package': title, 'version': version, })
return out
[docs] def get_additional_repository_information(self, repository_string):
"""
Function downloads additional information from quay.io to
get the tag-field which includes the version number.
"""
url = "{}/{}/{}".format(QUAY_API_URL, self.organization, repository_string)
r = requests.get(url, headers={'Accept-encoding': 'gzip'})
json_decoder = json.JSONDecoder()
decoded_request = json_decoder.decode(r.text)
return decoded_request['tags']
[docs]class CondaSearch():
"""
Tool to search the bioconda channel
"""
[docs] def get_json(self, search_string):
"""
Function takes search_string variable and returns results from the bioconda channel in JSON format
"""
if run_command is None:
raise Exception("Invalid search destination. " + deps_error_message("conda"))
raw_out, err, exit_code = run_command(
'search', '-c',
self.channel,
search_string,
use_exception_handler=True)
if exit_code != 0:
logging.info('Search failed with: %s' % err)
return []
return [{'package': n.split()[0], 'version': n.split()[1], 'build': n.split()[2]} for n in raw_out.split('\n')[2:-1]]
[docs]class GitHubSearch():
"""
Tool to search the GitHub bioconda-recipes repo
"""
[docs] def get_json(self, search_string):
"""
Takes search_string variable and return results from the bioconda-recipes github repository in JSON format
"""
response = requests.get(
"https://api.github.com/search/code?q=%s+in:path+repo:bioconda/bioconda-recipes+path:recipes" % search_string).json()
return response
[docs] def process_json(self, json, search_string):
"""
Take JSON input and process it, returning the required data
"""
json = json['items'][0:10] # get top ten results
results = []
for result in json:
results.append({'name': result['name'], 'path': result['path']})
return results
[docs] def recipe_present(self, search_string):
"""
Check if a recipe exists in bioconda-recipes which matches search_string exactly
"""
if requests.get("https://api.github.com/repos/bioconda/bioconda-recipes/contents/recipes/%s" % search_string).status_code == 200:
return True
else:
return False
[docs]def get_package_hash(packages, versions):
"""
Take packages and versions (if the latter are given) and returns a hash for each. Also checks github to see if the container is already present.
"""
hash_results = {}
targets = []
if versions:
for p in packages:
targets.append(build_target(p, version=versions[p]))
else: # if versions are not given only calculate the package hash
for p in packages:
targets.append(build_target(p))
# make the hash from the processed targets
package_hash = v2_image_name(targets)
hash_results['package_hash'] = package_hash.split(':')[0]
if versions:
hash_results['version_hash'] = package_hash.split(':')[1]
r = requests.get("https://quay.io/api/v1/repository/biocontainers/%s" % hash_results['package_hash'])
if r.status_code == 200:
hash_results['container_present'] = True
if versions: # now test if the version hash is listed in the repository tags
# remove -0, -1, etc from end of the tag
tags = [n[:-2] for n in r.json()['tags']]
if hash_results['version_hash'] in tags:
hash_results['container_present_with_version'] = True
else:
hash_results['container_present_with_version'] = False
else:
hash_results['container_present'] = False
return hash_results
[docs]def singularity_search(search_string):
"""
Check if a singularity package is present and return the link.
"""
results = []
containers = get_singularity_containers()
for container in containers:
if search_string in container:
name = container.split(':')[0]
version = container.split(':')[1]
results.append({'package': name, 'version': version})
return results
[docs]def readable_output(json, organization='biocontainers', channel='bioconda'):
# if json is empty:
if sum([len(json[destination][results]) for destination in json for results in json[destination]]) == 0:
sys.stdout.write('No results found for that query.\n')
return
# return results for quay, conda and singularity together
if sum([len(json[destination][results]) for destination in ['quay', 'conda', 'singularity', ] for results in json.get(destination, [])]) > 0:
sys.stdout.write("The query returned the following result(s).\n")
# put quay, conda etc results as lists in lines
lines = [['LOCATION', 'NAME', 'VERSION', 'COMMAND\n']]
for search_string, results in json.get('quay', {}).items():
for result in results:
lines.append(['quay', result['package'], result['version'], 'docker pull quay.io/%s/%s:%s\n' %
(organization, result['package'], result['version'])]) # NOT a real solution
for search_string, results in json.get('conda', {}).items():
for result in results:
lines.append(['conda', result['package'], '{}--{}'.format(result['version'], result['build']),
'conda install -c {} {}={}={}\n'.format(channel, result['package'], result['version'], result['build'])])
for search_string, results in json.get('singularity', {}).items():
for result in results:
lines.append(['singularity', result['package'], result['version'],
'wget https://depot.galaxyproject.org/singularity/{}:{}\n'.format(result['package'], result['version'])])
col_width0, col_width1, col_width2 = (max(len(
line[n]) for line in lines) + 2 for n in (0, 1, 2)) # def max col widths for the output
# create table
for line in lines:
sys.stdout.write("".join((line[0].ljust(col_width0), line[1].ljust(
col_width1), line[2].ljust(col_width2), line[3]))) # output
if json.get('github_recipe_present', False):
sys.stdout.write('\n' if 'lines' in locals() else '')
sys.stdout.write(
'The following recipes were found in the bioconda-recipes repository which exactly matched one of the search terms:\n')
lines = [['QUERY', 'LOCATION\n']]
for recipe in json['github_recipe_present']['recipes']:
lines.append(
[recipe, "https://api.github.com/repos/bioconda/bioconda-recipes/contents/recipes/%s\n" % recipe])
col_width0 = max(len(line[0]) for line in lines) + 2
for line in lines:
sys.stdout.write(
"".join((line[0].ljust(col_width0), line[1]))) # output
if sum([len(json['github'][results]) for results in json.get('github', [])]) > 0:
sys.stdout.write('\n' if 'lines' in locals() else '')
sys.stdout.write(
"Other result(s) on the bioconda-recipes GitHub repository:\n")
lines = [['QUERY', 'FILE', 'URL\n']]
for search_string, results in json.get('github', {}).items():
for result in results:
lines.append([search_string, result['name'],
'https://github.com/bioconda/bioconda-recipes/tree/master/%s\n' % result['path']])
# def max col widths for the output
col_width0, col_width1 = (
max(len(line[n]) for line in lines) + 2 for n in (0, 1))
for line in lines:
sys.stdout.write("".join(
(line[0].ljust(col_width0), line[1].ljust(col_width1), line[2]))) # output
[docs]def deps_error_message(package):
return "Required dependency [%s] is not installed. Run 'pip install galaxy-tool-util[mulled]'." % package
[docs]def main(argv=None):
if Schema is None:
sys.stdout.write(deps_error_message("Whoosh"))
return
destination_defaults = ['quay', 'singularity', 'github']
if run_command is not None:
destination_defaults.append('conda')
parser = argparse.ArgumentParser(
description='Searches in a given quay organization for a repository')
parser.add_argument('-d', '--destination', dest='search_dest', nargs='+', default=destination_defaults,
help="Choose where to search. Options are 'conda', 'quay', 'singularity' and 'github'. If no option are given, all will be searched.")
parser.add_argument('-o', '--organization', dest='organization_string', default="biocontainers",
help='Change quay organization to search; default is biocontainers.')
parser.add_argument('-c', '--channel', dest='channel_string', default="bioconda",
help='Change conda channel to search; default is bioconda.')
parser.add_argument('--non-strict', dest='non_strict', action="store_true",
help='Autocorrection of typos activated. Lists more results but can be confusing.\
For too many queries quay.io blocks the request and the results can be incomplete.')
parser.add_argument('-j', '--json', dest='json',
action="store_true", help='Returns results as JSON.')
parser.add_argument('-s', '--search', required=True, nargs='+',
help='The name of the tool(s) to search for.')
args = parser.parse_args()
json_results = {dest: {} for dest in args.search_dest}
versions = {}
if len(args.search) > 1: # get hash if multiple packages are searched
args.search.append(get_package_hash(
args.search, versions)['package_hash'])
if 'conda' in args.search_dest:
conda_results = {}
conda = CondaSearch(args.channel_string)
for item in args.search:
conda_results[item] = conda.get_json(item)
json_results['conda'] = conda_results
if 'github' in args.search_dest:
github_results = {}
github_recipe_present = []
github = GitHubSearch()
for item in args.search:
github_json = github.get_json(item)
github_results[item] = github.process_json(github_json, item)
if github.recipe_present(item):
github_recipe_present.append(item)
json_results['github'] = github_results
json_results['github_recipe_present'] = {
'recipes': github_recipe_present}
if 'quay' in args.search_dest:
quay_results = {}
quay = QuaySearch(args.organization_string)
quay.build_index()
for item in args.search:
quay_results[item] = quay.search_repository(item, args.non_strict)
json_results['quay'] = quay_results
if 'singularity' in args.search_dest:
singularity_results = {}
for item in args.search:
singularity_results[item] = singularity_search(item)
json_results['singularity'] = singularity_results
if args.json:
print(json_results)
else:
readable_output(json_results, args.organization_string,
args.channel_string)
if __name__ == "__main__":
main()