Source code for galaxy.tool_shed.tool_shed_registry

import logging
from typing import NamedTuple

from galaxy.util import parse_xml_string
from galaxy.util.tool_shed.common_util import remove_protocol_from_tool_shed_url
from galaxy.util.tool_shed.xml_util import parse_xml

log = logging.getLogger(__name__)

DEFAULT_TOOL_SHEDS_CONF_XML = """<?xml version="1.0"?>
<tool_sheds>
    <tool_shed name="Galaxy Main Tool Shed" url="https://toolshed.g2.bx.psu.edu/"/>
</tool_sheds>
"""


[docs]class AUTH_TUPLE(NamedTuple):
    username: str
    password: str


[docs]class Registry:

[docs]    def __init__(self, config=None):
        self.tool_sheds = {}
        self.tool_sheds_auth = {}
        if config:
            # Parse tool_sheds_conf.xml
            tree, error_message = parse_xml(config)
            if tree is None:
                log.warning(f"Unable to load references to tool sheds defined in file {str(config)}")
                return
            root = tree.getroot()
        else:
            root = parse_xml_string(DEFAULT_TOOL_SHEDS_CONF_XML)
            config = "internal default config"
        log.debug(f'Loading references to tool sheds from {config}')
        for elem in root.findall('tool_shed'):
            try:
                name = elem.get('name', None)
                url = elem.get('url', None)
                username = elem.get('user', None)
                password = elem.get('pass', None)
                if name and url:
                    self.tool_sheds[name] = url
                    self.tool_sheds_auth[name] = None
                    log.debug(f'Loaded reference to tool shed: {name}')
                if name and url and username and password:
                    self.tool_sheds_auth[name] = AUTH_TUPLE(username, password)
            except Exception as e:
                log.warning(f'Error loading reference to tool shed "{name}", problem: {str(e)}')

[docs]    def url_auth(self, url):
        """
        If the tool shed is using external auth, the client to the tool shed must authenticate to that
        as well.  This provides access to the six.moves.urllib.request.HTTPPasswordMgrWithdefaultRealm() object for the
        url passed in.

        Following more what galaxy.demo_sequencer.controllers.common does might be more appropriate at
        some stage...
        """
        url_sans_protocol = remove_protocol_from_tool_shed_url(url)
        for shed_name, shed_url in self.tool_sheds.items():
            shed_url_sans_protocol = remove_protocol_from_tool_shed_url(shed_url)
            if url_sans_protocol.startswith(shed_url_sans_protocol):
                return self.tool_sheds_auth[shed_name]
        log.debug(f"Invalid url '{str(url)}' received by tool shed registry's url_auth method.")
        return None