Source code for galaxy.webhooks

"""
This module manages loading of Galaxy webhooks.
"""

import logging
import os

import yaml

from galaxy.util import config_directories_from_setting

log = logging.getLogger(__name__)


[docs]class Webhook:
[docs]    def __init__(self, id, type, activate, weight, path):
        self.id = id
        self.type = type
        self.activate = activate
        self.weight = weight
        self.path = path
        self.styles = ""
        self.script = ""
        self.helper = ""
        self.config = {}

[docs]    def to_dict(self):
        return {
            "id": self.id,
            "type": self.type,
            "activate": self.activate,
            "weight": self.weight,
            "styles": self.styles,
            "script": self.script,
            "config": self.config,
        }


[docs]class WebhooksRegistry:
[docs]    def __init__(self, webhooks_dirs):
        self.webhooks = []
        self.webhooks_directories = []

        for webhook_dir in config_directories_from_setting(webhooks_dirs):
            for plugin_dir in os.listdir(webhook_dir):
                path = os.path.join(webhook_dir, plugin_dir)
                if os.path.isdir(path):
                    self.webhooks_directories.append(path)

        self.load_webhooks()

[docs]    def load_webhooks(self):
        for directory in self.webhooks_directories:
            config_file_path = None
            for config_file in ["config.yml", "config.yaml"]:
                path = os.path.join(directory, config_file)
                if os.path.isfile(path):
                    config_file_path = path
                    break

            if config_file_path:
                try:
                    self.load_webhook_from_config(directory, config_file_path)
                except Exception as e:
                    log.exception(e)

[docs]    def load_webhook_from_config(self, webhook_dir, config_file_path):
        with open(config_file_path) as fh:
            config = yaml.safe_load(fh)

        weight = config.get("weight", 1)
        if weight < 1:
            raise ValueError("Webhook weight must be greater or equal 1.")

        webhook = Webhook(
            config.get("id"),
            config.get("type"),
            config.get("activate", False),
            weight,
            webhook_dir,
        )

        # Read styles into a string, assuming all styles are in a
        # single file
        try:
            styles_file = os.path.join(webhook_dir, "styles.css")
            with open(styles_file) as fh:
                webhook.styles = fh.read().replace("\n", "")
        except OSError:
            pass

        # Read script into a string, assuming everything is in a
        # single file
        try:
            script_file = os.path.join(webhook_dir, "script.js")
            with open(script_file, encoding="utf-8") as fh:
                webhook.script = fh.read()
        except OSError:
            pass

        # Save helper function path if it exists
        helper_path = os.path.join(webhook_dir, "__init__.py")
        if os.path.isfile(helper_path):
            webhook.helper = helper_path

        webhook.config = config
        self.webhooks.append(webhook)