Source code for galaxy.managers.configuration

"""
Managers, serializers for Galaxy config file data. ConfigSerializer for all users
and a more expanded set of data for admin in AdminConfigSerializer.

Used by both the API and bootstrapped data.
"""
import json
import logging
import os
import sys
from typing import (
    Any,
    cast,
    Dict,
    List,
)

from galaxy.app import MinimalManagerApp
from galaxy.managers import base
from galaxy.managers.context import ProvidesUserContext
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.types import SerializationParams
from galaxy.web.framework.base import server_starttime

log = logging.getLogger(__name__)

VERSION_JSON_FILE = 'version.json'


[docs]class ConfigurationManager:
    """Interface/service object for interacting with configuration and related data."""

[docs]    def __init__(self, app: MinimalManagerApp):
        self._app = app

[docs]    def get_configuration(
        self,
        trans: ProvidesUserContext,
        serialization_params: SerializationParams
    ) -> Dict[str, Any]:
        is_admin = trans.user_is_admin
        serializer = AdminConfigSerializer(self._app) if is_admin else ConfigSerializer(self._app)
        return serializer.serialize_to_view(self._app.config, **serialization_params)

[docs]    def version(self) -> Dict[str, Any]:
        version_info = {
            "version_major": self._app.config.version_major,
            "version_minor": self._app.config.version_minor,
        }
        # Try loading extra version info
        json_file = os.path.join(self._app.config.root, VERSION_JSON_FILE)  # TODO: add this to schema
        json_file = os.environ.get("GALAXY_VERSION_JSON_FILE", json_file)
        try:
            with open(json_file) as f:
                extra_info = json.load(f)
        except OSError:
            log.info('Galaxy JSON version file not loaded')
        else:
            version_info['extra'] = extra_info
        return version_info

[docs]    def decode_id(
        self,
        encoded_id: EncodedDatabaseIdField,
    ) -> Dict[str, int]:
        # Handle the special case for library folders
        if ((len(encoded_id) % 16 == 1) and encoded_id.startswith('F')):
            encoded_id = cast(EncodedDatabaseIdField, encoded_id[1:])
        decoded_id = self._app.security.decode_id(encoded_id)
        return {"decoded_id": decoded_id}

[docs]    def tool_lineages(self) -> List[Dict[str, Dict]]:
        rval = []
        for id, tool in self._app.toolbox.tools():
            try:
                lineage_dict = tool.lineage.to_dict()
            except AttributeError:
                pass
            else:
                entry = {'id': id, 'lineage': lineage_dict}
                rval.append(entry)
        return rval

[docs]    def dynamic_tool_confs(self) -> List[Dict[str, str]]:
        # WARNING: If this method is ever changed so as not to require admin privileges, update the nginx proxy
        # documentation, since this path is used as an authentication-by-proxy method for securing other paths on the
        # server. A dedicated endpoint should probably be added to do that instead.
        def tool_conf_to_dict(conf):
            return dict(
                config_filename=conf['config_filename'],
                tool_path=conf['tool_path'],
            )
        confs = self._app.toolbox.dynamic_confs(include_migrated_tool_conf=True)
        return list(map(tool_conf_to_dict, confs))

[docs]    def reload_toolbox(self):
        self._app.queue_worker.send_control_task('reload_toolbox')


# TODO: this is a bit of an odd duck. It uses the serializer structure from managers
#   but doesn't have a model like them. It might be better in config.py or a
#   totally new area, but I'm leaving it in managers for now for class consistency.
[docs]class ConfigSerializer(base.ModelSerializer):
    """Configuration (galaxy.ini) settings viewable by all users"""

[docs]    def __init__(self, app):
        super().__init__(app)

        self.default_view = 'all'
        self.add_view('all', list(self.serializers.keys()))

[docs]    def default_serializer(self, config, key):
        return getattr(config, key, None)

[docs]    def add_serializers(self):

        def _defaults_to(default):
            return lambda config, key, **context: getattr(config, key, default)

        def _use_config(config, key, **context):
            """Let config object determine the value for key"""
            assert hasattr(config, key)
            return getattr(config, key)

        def _config_is_truthy(config, key, **context):
            return True if config.get(key) else False

        self.serializers = {
            # TODO: this is available from user data, remove
            'is_admin_user': lambda *a, **c: False,
            'brand': _use_config,
            'display_galaxy_brand': _use_config,
            'logo_url': _use_config,
            'logo_src': _use_config,
            'logo_src_secondary': _use_config,
            'terms_url': _use_config,
            'myexperiment_target_url': _use_config,
            'wiki_url': _use_config,
            'search_url': _use_config,
            'mailing_lists': _defaults_to(self.app.config.mailing_lists_url),
            'screencasts_url': _use_config,
            'citation_url': _use_config,
            'support_url': _use_config,
            'quota_url': _use_config,
            'helpsite_url': _use_config,
            'lims_doc_url': _defaults_to("https://usegalaxy.org/u/rkchak/p/sts"),
            'default_locale': _use_config,
            'enable_openid': _use_config,
            'enable_tool_recommendations': _use_config,
            'tool_recommendation_model_path': _use_config,
            'admin_tool_recommendations_path': _use_config,
            'overwrite_model_recommendations': _use_config,
            'topk_recommendations': _use_config,
            'allow_user_impersonation': _use_config,
            'allow_user_creation': _defaults_to(False),  # schema default is True
            'use_remote_user': _defaults_to(None),  # schema default is False; or config.single_user
            'single_user': _config_is_truthy,
            'enable_oidc': _use_config,
            'oidc': _use_config,
            'enable_quotas': _use_config,
            'remote_user_logout_href': _use_config,
            'datatypes_disable_auto': _use_config,
            'allow_user_dataset_purge': _defaults_to(False),  # schema default is True
            'ga_code': _use_config,
            'plausible_server': _use_config,
            'plausible_domain': _use_config,
            'matomo_server': _use_config,
            'matomo_site_id': _use_config,
            'enable_unique_workflow_defaults': _use_config,
            'enable_beta_markdown_export': _use_config,
            'simplified_workflow_run_ui': _use_config,
            'simplified_workflow_run_ui_target_history': _use_config,
            'simplified_workflow_run_ui_job_cache': _use_config,
            'has_user_tool_filters': _defaults_to(False),
            # TODO: is there no 'correct' way to get an api url? controller='api', action='tools' is a hack
            # at any rate: the following works with path_prefix but is still brittle
            # TODO: change this to (more generic) upload_path and incorporate config.nginx_upload_path into building it
            'nginx_upload_path': lambda config, key, **context: getattr(config, key, False),
            'chunk_upload_size': _use_config,
            'ftp_upload_site': _use_config,
            'version_major': _defaults_to(None),
            'version_minor': _defaults_to(None),
            'require_login': _use_config,
            'inactivity_box_content': _use_config,
            'visualizations_visible': _use_config,
            'interactivetools_enable': _use_config,
            'aws_estimate': _use_config,
            'message_box_content': _use_config,
            'message_box_visible': _use_config,
            'message_box_class': _use_config,
            'server_startttime': lambda config, key, **context: server_starttime,
            'mailing_join_addr': _defaults_to('galaxy-announce-join@bx.psu.edu'),  # should this be the schema default?
            'server_mail_configured': lambda config, key, **context: bool(config.smtp_server),
            'registration_warning_message': _use_config,
            'welcome_url': _use_config,
            'show_welcome_with_login': _defaults_to(True),  # schema default is False
            'cookie_domain': _use_config,
            'python': _defaults_to((sys.version_info.major, sys.version_info.minor)),
            'select_type_workflow_threshold': _use_config,
            'file_sources_configured': lambda config, key, **context: self.app.file_sources.custom_sources_configured,
            'upload_from_form_button': _use_config,
            'release_doc_base_url': _use_config,
            'user_library_import_dir_available': lambda config, key, **context: bool(config.get('user_library_import_dir')),
        }


[docs]class AdminConfigSerializer(ConfigSerializer):
    """Configuration attributes viewable only by admin users"""

[docs]    def add_serializers(self):
        super().add_serializers()

        def _defaults_to(default):
            return lambda config, key, **context: getattr(config, key, default)

        self.serializers.update({
            # TODO: this is available from user serialization: remove
            'is_admin_user': lambda *a: True,

            'library_import_dir': _defaults_to(None),
            'user_library_import_dir': _defaults_to(None),
            'allow_library_path_paste': _defaults_to(False),
            'allow_user_deletion': _defaults_to(False),
        })