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.tools.deps.mulled.mulled_search
#!/usr/bin/env python
import argparse
import json
import sys
import tempfile
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:
search_string = "*%s*" % search_string
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)
sys.stdout.write("The query \033[1m %s \033[0m resulted in %s result(s).\n" % (search_string, len(results)))
if non_strict:
sys.stdout.write('The search was relaxed and the following search terms were searched: ')
sys.stdout.write('\033[1m %s \033[0m\n' % ', '.join(suggestions))
out = list()
for result in results:
title = result['title']
for version in self.get_additional_repository_information(title):
row = [title]
row.append(version)
out.append(row)
if out:
col_width = max(len(word) for row in out for word in row) + 2 # padding
for row in out:
name = row[0]
version = row[1]
sys.stdout.write("".join(word.ljust(col_width) for word in row) + "docker pull quay.io/%s/%s:%s\n" % (self.organization, name, version))
else:
sys.stdout.write("No results found for %s in quay.io/%s.\n" % (search_string, self.organization))
[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 = "%s/%s/%s" % (QUAY_API_URL, self.organization, repository_string)
r = requests.get(url, headers={'Accept-encoding': 'gzip'}, timeout=12)
json_decoder = json.JSONDecoder()
decoded_request = json_decoder.decode(r.text)
return decoded_request['tags']
[docs]def main(argv=None):
parser = argparse.ArgumentParser(description='Searches in a given quay organization for a repository')
parser.add_argument('-o', '--organization', dest='organization_string', default="biocontainers",
help='Change organization. Default is biocontainers.')
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('-s', '--search', required=True,
help='The name of the tool you want to search for.')
args = parser.parse_args()
quay = QuaySearch(args.organization_string)
quay.build_index()
quay.search_repository(args.search, args.non_strict)
if __name__ == "__main__":
main()