Source code for galaxy.tool_util.deps.mulled.mulled_list

#!/usr/bin/env python

import argparse
import logging
from glob import glob
from html.parser import HTMLParser

from galaxy.util import requests
from .util import MULLED_SOCKET_TIMEOUT

QUAY_API_ENDPOINT = "https://quay.io/api/v1/repository"


[docs]def get_quay_containers(repository="biocontainers"): """ Get all quay containers in the biocontainers repo """ containers = [] repos_parameters = {"public": "true", "namespace": repository} repos_headers = {"Accept-encoding": "gzip", "Accept": "application/json"} repos_response = requests.get(QUAY_API_ENDPOINT, headers=repos_headers, params=repos_parameters, timeout=12) repos = repos_response.json()["repositories"] for repo in repos: logging.info(repo) tags_response = requests.get(f"{QUAY_API_ENDPOINT}/{repository}/{repo['name']}", timeout=MULLED_SOCKET_TIMEOUT) tags = tags_response.json()["tags"] for tag in tags: containers.append(f"{repo['name']}:{tag}") return containers
[docs]def get_singularity_containers(): """ Get all existing singularity containers from "https://depot.galaxyproject.org/singularity/" """ class GetContainerNames(HTMLParser): # small parser which gets list of containers def __init__(self): HTMLParser.__init__(self) self.containers = [] def handle_starttag(self, tag, attrs): try: for attr in attrs: if attr[0] == "href" and attr[1] != "../": self.containers.append(attr[1].replace("%3A", ":")) except IndexError: pass parser = GetContainerNames() index = requests.get("https://depot.galaxyproject.org/singularity/", timeout=MULLED_SOCKET_TIMEOUT) parser.feed(index.text) return parser.containers
[docs]def get_conda_envs(filepath): """ Get list of already existing envs """ return [n.split("__")[-1].replace("@", ":") for n in glob(f"{filepath}/*")]
[docs]def get_missing_containers(quay_list, singularity_list, blocklist_file=None): r""" Return list of quay containers that do not exist as singularity containers. Files stored in a blocklist will be ignored """ blocklist = [] if blocklist_file: with open(blocklist_file) as fh: blocklist = fh.read().split("\n") return [n for n in quay_list if n not in singularity_list and n not in blocklist]
[docs]def get_missing_envs(quay_list, conda_list, blocklist_file=None): r""" Compares list of conda envs and docker containers and returns missing conda envs """ blocklist = [] if blocklist_file: with open(blocklist_file) as fh: blocklist = fh.read().split("\n") return [n for n in quay_list if n.split("--")[0] not in conda_list and n.split("--")[0] not in blocklist]
[docs]def main(): parser = argparse.ArgumentParser( description="Returns list of Docker containers in the quay.io biocontainers repository." ) parser.add_argument("--source", "-s", help="Docker, Singularity or Conda.") parser.add_argument( "--not-singularity", dest="not_singularity", action="store_true", help="Exclude Docker containers from which Singularity containers have already been built.", ) parser.add_argument( "--not-conda", dest="not_conda", action="store_true", help="Exclude Docker containers from which Conda environments have already been extracted.", ) parser.add_argument( "--conda-filepath", dest="conda_filepath", default=None, help="If searching for conda environments or employing the --not-conda option, a filepath where the environments are located.", ) parser.add_argument( "-b", "--blocklist", "--blacklist", dest="blocklist", default=None, help="Provide a 'blocklist file' containing containers which should be excluded from the list.", ) parser.add_argument( "-f", "--file", dest="output", default=None, help="File to write list to. If not given output will be returned on the command line.", ) args = parser.parse_args() if args.source == "docker": containers = get_quay_containers() if args.not_singularity: containers = get_missing_containers(containers, get_singularity_containers(), args.blocklist) if args.not_conda: containers = get_missing_envs(containers, get_conda_envs(args.conda_filepath), args.blocklist) elif args.source == "singularity": containers = get_singularity_containers() elif args.source == "conda": containers = get_conda_envs(args.conda_filepath) else: print("The 'source' argument was not understood.") return if args.output: with open(args.output, "a") as f: for container in containers: f.write(f"{container}\n") else: print(containers)
if __name__ == "__main__": main()