Source code for galaxy.dependencies

"""
Determine what optional dependencies are needed.
"""

import logging
import os
import re
import sys
from collections.abc import Mapping
from os.path import (
    dirname,
    exists,
    join,
)
from typing import (
    Any,
    cast,
)

import yaml
from dparse import parse
from packaging.requirements import Requirement

from galaxy.config import GalaxyAppConfiguration
from galaxy.util import (
    asbool,
    etree,
    parse_xml,
    which,
)
from galaxy.util.config_templates import apply_syntactic_sugar
from galaxy.util.properties import (
    find_config_file,
    load_app_properties,
)

log = logging.getLogger(__name__)

# The cloud object store's provider SDKs are cloudbridge extras, so only the
# ones backing a configured provider need installing. Galaxy's provider name
# matches the extra except for Google.
CLOUDBRIDGE_EXTRAS = {
    "aws": "aws",
    "azure": "azure",
    "google": "gcp",
    "openstack": "openstack",
}



[docs]
class BaseConditionalDependencies:
    """Machinery and checks shared by every app that ships a config file.

    A ``check_<dependency>`` method here must only read options that every
    subclass' config schema defines. App-specific checks belong on the subclass;
    ``check()`` treats a missing method as "not needed".
    """

    # Section to read from a YAML config file; None keeps load_app_properties'
    # own default, which is what the Galaxy config path has always relied on.
    config_section: str | None = None


[docs]
    def __init__(self, config_file, config=None):
        self.config_file = config_file
        self.conditional_reqs = []
        if config is None:
            self.config = load_app_properties(config_file=self.config_file, config_section=self.config_section)
        else:
            self.config = config
        self.parse_configs()
        self.get_conditional_requirements()



[docs]
    def parse_configs(self):
        """Collect app-specific signals from auxiliary config files."""



[docs]
    def get_conditional_requirements(self):
        crfile = join(dirname(__file__), "conditional-requirements.txt")
        with open(crfile) as fh:
            dependency_file = parse(fh.read(), file_type="requirements.txt")
            for dep in dependency_file.dependencies:
                self.conditional_reqs.append(dep)



[docs]
    def check(self, name):
        try:
            name = name.replace("-", "_").replace(".", "_")
            return getattr(self, f"check_{name}")()
        except Exception:
            return False



[docs]
    def extras(self, name: str) -> list[str]:
        """Extras of the named package that this configuration requires."""
        name = name.replace("-", "_").replace(".", "_")
        hook = getattr(self, f"extras_{name}", None)
        return hook() if hook else []



[docs]
    def check_psycopg2_binary(self):
        return self.config["database_connection"].startswith(("postgresql://", "postgresql+psycopg2://"))



[docs]
    def check_psycopg(self):
        return self.config["database_connection"].startswith("postgresql+psycopg://")



[docs]
    def check_mysqlclient(self):
        return self.config["database_connection"].startswith("mysql")



[docs]
    def check_sentry_sdk(self):
        return self.config.get("sentry_dsn", None) is not None





[docs]
class ToolShedConditionalDependencies(BaseConditionalDependencies):
    config_section = "tool_shed"




[docs]
class ConditionalDependencies(BaseConditionalDependencies):

[docs]
    def __init__(self, config_file, config=None):
        self.job_runners = []
        self.authenticators = []
        self.object_stores = []
        self.cloud_object_store_providers = []
        self.file_sources = []
        self.container_interface_types = []
        self.job_rule_modules = []
        self.error_report_modules = []
        self.vault_type = None
        super().__init__(config_file, config=config)



[docs]
    def parse_configs(self):
        self.config_object = GalaxyAppConfiguration(config_file=self.config_file, override_tempdir=False, **self.config)

        def load_job_config_dict(job_conf_dict):
            runners = job_conf_dict.get("runners", {})
            for runner in runners.values():
                if "load" in runner:
                    self.job_runners.append(runner.get("load"))
            environments = job_conf_dict.get("execution", {}).get("environments", {})
            for env in environments.values():
                runner = env.get("runner")
                if runner == "dynamic_tpv":
                    self.job_rule_modules.append("tpv.rules")
                elif "rules_module" in env:
                    self.job_rule_modules.append(env.get("rules_module"))

        if "job_config" in self.config:
            load_job_config_dict(self.config.get("job_config"))
        else:
            job_conf_path = self.config_object.job_config_file
            if not job_conf_path:
                job_conf_path = join(dirname(self.config_file), "job_conf.yml")
                if not exists(job_conf_path):
                    job_conf_path = join(dirname(self.config_file), "job_conf.xml")
            else:
                job_conf_path = join(dirname(self.config_file), job_conf_path)
            if ".xml" in job_conf_path:
                try:
                    job_conf_tree = parse_xml(job_conf_path)
                    plugins_elem = job_conf_tree.find("plugins")
                    if plugins_elem:
                        for plugin in plugins_elem.findall("plugin"):
                            if "load" in plugin.attrib:
                                self.job_runners.append(plugin.attrib["load"])
                    for plugin in job_conf_tree.findall('.//destination/param[@id="rules_module"]'):
                        self.job_rule_modules.append(plugin.text)
                except (OSError, etree.ParseError):
                    pass
            else:
                try:
                    with open(job_conf_path) as f:
                        job_conf_dict = yaml.safe_load(f)
                    load_job_config_dict(job_conf_dict)
                except OSError:
                    pass

        object_store_conf_path = self.config_object.object_store_config_file
        try:
            if ".xml" in object_store_conf_path:
                for store in parse_xml(object_store_conf_path).iter("object_store"):
                    if "type" in store.attrib:
                        # lxml's _Attrib is str-keyed in practice but not typed as a Mapping.
                        self.collect_object_store(cast(Mapping[str, Any], store.attrib))
            else:
                with open(object_store_conf_path) as f:
                    job_conf_dict = yaml.safe_load(f)

                def collect_types(from_dict):
                    if not isinstance(from_dict, dict):
                        return

                    if "type" in from_dict:
                        self.collect_object_store(from_dict)

                    for value in from_dict.values():
                        if isinstance(value, list):
                            for val in value:
                                collect_types(val)
                        else:
                            collect_types(value)

                collect_types(job_conf_dict)

        except OSError:
            pass

        # Parse object store templates config. Stores offered as templates need
        # their dependencies as much as ones configured in object_store_conf.
        try:
            object_store_templates_conf = self.config.get("object_store_templates")
            if object_store_templates_conf is None:
                object_store_templates_conf_yml = self.config_object.object_store_templates_config_file
                if object_store_templates_conf_yml and exists(object_store_templates_conf_yml):
                    with open(object_store_templates_conf_yml) as f:
                        object_store_templates_conf = yaml.safe_load(f)
            for object_store_template in apply_syntactic_sugar(object_store_templates_conf or []):
                configuration = object_store_template.get("configuration") or {}
                if "type" in configuration:
                    self.collect_object_store(configuration)
        except OSError:
            pass

        # Parse auth conf
        auth_conf_xml = self.config_object.auth_config_file
        try:
            for auth in parse_xml(auth_conf_xml).findall("authenticator"):
                auth_type = auth.find("type")
                if auth_type is not None:
                    self.authenticators.append(auth_type.text)
        except OSError:
            pass

        # Install pkce whenever OIDC is in use. PKCE can be toggled per-provider
        # in oidc_backends_config.xml at runtime, so tying the install decision
        # to any top-level OIDC signal avoids rebuilding the venv when
        # pkce_support is flipped (issue #22502).
        self.oidc_active = (
            asbool(self.config.get("enable_oidc", False))
            or bool(self.config.get("oidc_auth_pipeline"))
            or bool(self.config.get("oidc_auth_pipeline_extra"))
        )

        # Parse error report config
        error_report_yml = self.config_object.error_report_file
        try:
            with open(error_report_yml) as f:
                error_reporters = yaml.safe_load(f)
                self.error_report_modules = [er.get("type", None) for er in error_reporters]
        except OSError:
            pass

        # Parse file sources config
        file_sources_conf_yml = self.config_object.file_sources_config_file
        if exists(file_sources_conf_yml):
            with open(file_sources_conf_yml) as f:
                file_sources_conf = yaml.safe_load(f)
        else:
            file_sources_conf = []
        self.file_sources = [c.get("type", None) for c in file_sources_conf]

        # Parse file source templates config
        file_source_templates_conf_yml = self.config_object.file_source_templates_config_file
        if file_source_templates_conf_yml and exists(file_source_templates_conf_yml):
            with open(file_source_templates_conf_yml) as f:
                file_source_templates_conf = apply_syntactic_sugar(yaml.safe_load(f))
            for file_source_template in file_source_templates_conf:
                self.file_sources.append(file_source_template["configuration"].get("type"))

        # Parse vault config
        vault_conf_yml = self.config_object.vault_config_file
        if exists(vault_conf_yml):
            with open(vault_conf_yml) as f:
                vault_conf = yaml.safe_load(f)
        else:
            vault_conf = {}
        self.vault_type = vault_conf.get("type", "").lower()



[docs]
    def collect_object_store(self, store: Mapping[str, Any]) -> None:
        """Record a configured object store, given its XML attributes or config dict."""
        self.object_stores.append(store["type"])
        if store["type"] == "cloud" and store.get("provider"):
            provider = str(store["provider"]).lower()
            if provider not in CLOUDBRIDGE_EXTRAS:
                log.warning(
                    "Unknown cloud object store provider '%s'; no cloudbridge extra will be installed for it.",
                    provider,
                )
            self.cloud_object_store_providers.append(provider)



[docs]
    def check_drmaa(self):
        return (
            "galaxy.jobs.runners.drmaa:DRMAAJobRunner" in self.job_runners
            or "galaxy.jobs.runners.slurm:SlurmJobRunner" in self.job_runners
            or "galaxy.jobs.runners.univa:UnivaJobRunner" in self.job_runners
        )



[docs]
    def check_galaxycloudrunner(self):
        return "galaxycloudrunner.rules" in self.job_rule_modules



[docs]
    def check_total_perspective_vortex(self):
        return "tpv.rules" in self.job_rule_modules



[docs]
    def check_pbs_python(self):
        return "galaxy.jobs.runners.pbs:PBSJobRunner" in self.job_runners



[docs]
    def check_pykube_ng(self):
        return "galaxy.jobs.runners.kubernetes:KubernetesJobRunner" in self.job_runners or which("kubectl")



[docs]
    def check_chronos_python(self):
        return "galaxy.jobs.runners.chronos:ChronosJobRunner" in self.job_runners



[docs]
    def check_htcondor(self):
        return (
            "galaxy.jobs.runners.htcondor:HTCondorJobRunner" in self.job_runners
            or os.environ.get("GALAXY_DEPENDENCIES_INSTALL_HTCONDOR") == "1"
        )



[docs]
    def check_boto3_python(self):
        return "galaxy.jobs.runners.aws:AWSBatchJobRunner" in self.job_runners



[docs]
    def check_fluent_logger(self):
        return asbool(self.config["fluent_log"])



[docs]
    def check_statsd(self):
        return self.config.get("statsd_host", None) is not None



[docs]
    def check_python_ldap(self):
        return "ldap" in self.authenticators or "activedirectory" in self.authenticators



[docs]
    def check_ldap3(self):
        return "ldap3" in self.authenticators



[docs]
    def check_python_pam(self):
        return "PAM" in self.authenticators



[docs]
    def check_azure_storage(self):
        return "azure_blob" in self.object_stores



[docs]
    def check_boto3(self):
        return "boto3" in self.object_stores



[docs]
    def check_cloudbridge(self):
        return "cloud" in self.object_stores



[docs]
    def extras_cloudbridge(self) -> list[str]:
        return sorted(
            {
                CLOUDBRIDGE_EXTRAS[provider]
                for provider in self.cloud_object_store_providers
                if provider in CLOUDBRIDGE_EXTRAS
            }
        )



[docs]
    def check_kamaki(self):
        return "pithos" in self.object_stores



[docs]
    def check_python_irodsclient(self):
        return "irods" in self.object_stores



[docs]
    def check_dropboxdrivefs(self):
        return "dropbox" in self.file_sources



[docs]
    def check_webdav4(self):
        return "webdav" in self.file_sources



[docs]
    def check_fs_anvilfs(self):
        # pyfilesystem plugin access to terra on anvil
        return "anvil" in self.file_sources



[docs]
    def check_fs_sshfs(self):
        return "ssh" in self.file_sources



[docs]
    def check_gdrive_fsspec(self):
        return "googledrive" in self.file_sources



[docs]
    def check_gcsfs(self):
        return "googlecloudstorage" in self.file_sources



[docs]
    def check_onedatafilerestclient(self):
        return "onedata" in self.object_stores



[docs]
    def check_fs_onedatarestfs(self):
        return "onedata" in self.file_sources



[docs]
    def check_fs_basespace(self):
        return "basespace" in self.file_sources



[docs]
    def check_rspace_client(self):
        return "rspace" in self.file_sources



[docs]
    def check_fs_irods(self):
        return "irods" in self.file_sources



[docs]
    def check_pybis(self):
        return "openbis" in self.file_sources



[docs]
    def check_watchdog(self):
        install_set = {"auto", "True", "true", "polling", True}
        return self.config["watch_tools"] in install_set or self.config["watch_tool_data_dir"] in install_set



[docs]
    def check_python_gitlab(self):
        return "gitlab" in self.error_report_modules



[docs]
    def check_pygithub(self):
        return "github" in self.error_report_modules



[docs]
    def check_influxdb(self):
        return "influxdb" in self.error_report_modules



[docs]
    def check_tensorflow(self):
        return asbool(self.config["enable_tool_recommendations"])



[docs]
    def check_openai(self):
        return self.config.get("openai_api_key", None) is not None



[docs]
    def check_pydantic_ai(self):
        return (
            self.config.get("ai_api_key", None) is not None or self.config.get("inference_services", None) is not None
        )



[docs]
    def check_weasyprint(self):
        # See notes in ./conditional-requirements.txt for more information.
        return os.environ.get("GALAXY_DEPENDENCIES_INSTALL_WEASYPRINT") == "1"



[docs]
    def check_pydyf(self):
        # See notes in ./conditional-requirements.txt for more information.
        return os.environ.get("GALAXY_DEPENDENCIES_INSTALL_WEASYPRINT") == "1"



[docs]
    def check_hvac(self):
        return "hashicorp" == self.vault_type



[docs]
    def check_pkce(self):
        return self.oidc_active



[docs]
    def check_rucio_clients(self):
        return "rucio" in self.object_stores



[docs]
    def check_redis(self):
        celery_enabled = self.config.get("enable_celery_tasks", False)
        celery_conf = self.config.get("celery_conf") or {}
        celery_result_backend = celery_conf.get("result_backend") or ""
        celery_broker_url = celery_conf.get("broker_url") or ""

        def is_redis_url(url: str) -> bool:
            # https://docs.celeryq.dev/en/stable/userguide/configuration.html#conf-redis-result-backend
            protocol = url.split("://")[0]
            return protocol in {"redis", "rediss", "redis+socket", "socket"}

        return celery_enabled and is_redis_url(celery_result_backend) or is_redis_url(celery_broker_url)



[docs]
    def check_adlfs(self):
        return "azureflat" in self.file_sources



[docs]
    def check_huggingface_hub(self):
        return "huggingface" in self.file_sources



[docs]
    def check_omero_py(self):
        return "omero" in self.file_sources



[docs]
    def check_iiif_fsspec(self):
        return "iiif" in self.file_sources



[docs]
    def check_fsspec_xrootd(self):
        return "xrootd" in self.file_sources



[docs]
    def check_ipfsspec(self):
        return "ipfs" in self.file_sources



[docs]
    def check_mavedb_fsspec(self):
        return "mavedb" in self.file_sources



[docs]
    def check_commoncrawl_fsspec(self):
        return "commoncrawl" in self.file_sources



[docs]
    def check_arcfs_fsspec(self):
        return "gitlab" in self.file_sources or "arc" in self.file_sources





[docs]
def strip_comment(line):
    # lifted from https://github.com/tox-dev/tox/commit/3c6b4f204e89852c4b7536b246a66d20be6d39ec
    # xref https://github.com/pyupio/dparse/issues/34
    return re.sub(r"\s+#.*", "", line).strip()



GALAXY_APP = "galaxy"
TOOL_SHED_APP = "tool_shed"

# Each app resolves its own config file and evaluates its own set of checks, so
# that conditional-requirements.txt governs every app we ship, not just Galaxy.
APPS = {
    GALAXY_APP: (ConditionalDependencies, ["galaxy", "universe_wsgi"]),
    TOOL_SHED_APP: (ToolShedConditionalDependencies, ["tool_shed", "tool_shed_wsgi"]),
}



[docs]
def optional(config_file=None, app=GALAXY_APP):
    try:
        dependencies_class, config_file_names = APPS[app]
    except KeyError:
        raise ValueError(f"Unknown app '{app}', expected one of: {', '.join(sorted(APPS))}")
    if not config_file:
        config_file = find_config_file(config_file_names, include_samples=True)
    if not config_file:
        print(f"galaxy.dependencies.optional: no {app} config file found", file=sys.stderr)
        return []
    rval = []
    conditional = dependencies_class(config_file)
    for dependency in conditional.conditional_reqs:
        if conditional.check(dependency.name):
            line = strip_comment(dependency.line)
            extras = conditional.extras(dependency.name)
            if extras:
                requirement = Requirement(line)
                requirement.extras |= set(extras)
                line = str(requirement)
            rval.append(line)
    return rval