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): = 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":, "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 ="\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 = except OSError: pass # Save helper function path if it exists helper_path = os.path.join(webhook_dir, "") if os.path.isfile(helper_path): webhook.helper = helper_path webhook.config = config self.webhooks.append(webhook)