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.tools.deps.mulled.mulled_search

#!/usr/bin/env python

import argparse
import json
import sys
import tempfile

    import requests
except ImportError:
    requests = None

    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 __init__(self, organization): self.index = None self.organization = organization
[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()