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

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()