Source code for galaxy.tours._impl

"""
This module manages loading/etc of Galaxy interactive tours.
"""

import logging
import os
from typing import (
    List,
    Union,
)

import yaml
from pydantic import parse_obj_as

from galaxy.exceptions import ObjectNotFound
from galaxy.navigation.data import load_root_component
from galaxy.util import config_directories_from_setting
from ._interface import ToursRegistry
from ._schema import TourList

log = logging.getLogger(__name__)

TOUR_EXTENSIONS = (".yml", ".yaml")

ROOT_COMPONENT = load_root_component()


[docs]def build_tours_registry(tour_directories: str): return ToursRegistryImpl(tour_directories)
def noop_warn(str): pass def load_tour_steps(contents_dict, warn=None, resolve_components=True): warn = warn or noop_warn # Some of this can be done on the clientside. Maybe even should? title_default = contents_dict.get("title_default") if "requirements" not in contents_dict: contents_dict["requirements"] = [] for step in contents_dict["steps"]: # Remove attributes no longer used, so they are attempted to be # validated. if "backdrop" in step: warn(f"Deprecated and dropped property backdrop found in step {step}") step.pop("backdrop") if "component" in step and resolve_components: component = step.pop("component") step["element"] = ROOT_COMPONENT.resolve_component_locator(component).locator if "intro" in step: step["content"] = step.pop("intro") if "position" in step: step["placement"] = step.pop("position") if "element" not in step: step["orphan"] = True if title_default and "title" not in step: step["title"] = title_default def get_tour_id_from_path(tour_path: Union[str, os.PathLike]) -> str: filename = os.path.basename(tour_path) return os.path.splitext(filename)[0] def load_tour_from_path(tour_path: Union[str, os.PathLike], warn=None, resolve_components=True) -> dict: with open(tour_path) as f: tour = yaml.safe_load(f) load_tour_steps(tour, warn=warn, resolve_components=resolve_components) return tour def is_yaml(filename: str) -> bool: for ext in TOUR_EXTENSIONS: if filename.endswith(ext): return True return False def tour_paths(target_path: Union[str, os.PathLike]) -> List[str]: paths = [] if os.path.isdir(target_path): for filename in os.listdir(target_path): if is_yaml(filename): paths.append(str(os.path.join(target_path, filename))) else: paths.append(str(target_path)) return paths @ToursRegistry.register class ToursRegistryImpl: def __init__(self, tour_directories): self.tour_directories = config_directories_from_setting(tour_directories) self._load_tours() def get_tours(self): """Return list of tours.""" tours = [] for k in self.tours.keys(): tourdata = { "id": k, "name": self.tours[k].get("name"), "description": self.tours[k].get("description"), "tags": self.tours[k].get("tags"), "requirements": self.tours[k].get("requirements"), } tours.append(tourdata) return parse_obj_as(TourList, tours) def tour_contents(self, tour_id): """Return tour contents.""" # Extra format translation could happen here (like the previous intro_to_tour) # For now just return the loaded contents. if tour_id not in self.tours: raise ObjectNotFound(f"tour {tour_id} not found") return self.tours.get(tour_id) def load_tour(self, tour_id): """Reload tour and return its contents.""" tour_path = self._get_path_from_tour_id(tour_id) self._load_tour_from_path(tour_path) return self.tours.get(tour_id) def reload_tour(self, path): """Reload tour.""" # We may safely assume that the path is within the tour directory filename = os.path.basename(path) if is_yaml(filename): self._load_tour_from_path(path) def _load_tours(self): self.tours = {} for tour_dir in self.tour_directories: for tour_path in tour_paths(tour_dir): self._load_tour_from_path(tour_path) def _load_tour_from_path(self, tour_path): tour_id = get_tour_id_from_path(tour_path) try: tour = load_tour_from_path(tour_path) except OSError: log.exception(f"Tour '{tour_id}' could not be loaded, error reading file.") except yaml.error.YAMLError: log.exception(f"Tour '{tour_id}' could not be loaded, error within file. Please check your YAML syntax.") except TypeError: log.exception( f"Tour '{tour_id}' could not be loaded, error within file." " Possibly spacing related. Please check your YAML syntax." ) self.tours[tour_id] = tour log.info(f"Loaded tour '{tour_id}'") def _get_path_from_tour_id(self, tour_id): for tour_dir in self.tour_directories: for ext in TOUR_EXTENSIONS: tour_path = os.path.join(tour_dir, tour_id + ext) if os.path.exists(tour_path): return tour_path