Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.web.base.pluginframework

"""
Base class for plugins - frameworks or systems that may:
 * add code at startup
 * allow hooks to be called
and base class for plugins that:
 * serve static content
 * serve templated html
 * have some configuration at startup
"""
import imp
import logging
import os.path
import sys

import mako.lookup

from galaxy import util
from galaxy.util import (
    bunch,
    odict
)

log = logging.getLogger(__name__)


[docs]class PluginManagerException(Exception): """Base exception for plugin frameworks. """ pass
[docs]class PluginManagerConfigException(PluginManagerException): """Exception for plugin framework configuration errors. """ pass
# ============================================================================= base
[docs]class PluginManager(object): """ Plugins represents an section of code that is not tracked in the Galaxy repository, allowing the addition of custom code to a Galaxy installation without changing the code base. A PluginManager discovers and manages these plugins. This is an non-abstract class but its usefulness is limited and is meant to be inherited. """
[docs] def __init__(self, app, directories_setting=None, skip_bad_plugins=True, **kwargs): """ Set up the manager and load all plugins. :type app: UniverseApplication :param app: the application (and its configuration) using this manager :type directories_setting: string (default: None) :param directories_setting: the filesystem path (or paths) to search for plugins. Can be CSV string of paths. Will be treated as absolute if a path starts with '/', relative otherwise. :type skip_bad_plugins: boolean (default: True) :param skip_bad_plugins: whether to skip plugins that cause exceptions when loaded or to raise that exception """ self.directories = [] self.skip_bad_plugins = skip_bad_plugins self.plugins = odict.odict() self.directories = util.config_directories_from_setting(directories_setting, app.config.root) self.load_configuration() self.load_plugins()
[docs] def load_configuration(self): """ Override to load some framework/plugin specifc configuration. """ # Abstract method return True
[docs] def load_plugins(self): """ Search ``self.directories`` for potential plugins, load them, and cache in ``self.plugins``. :rtype: odict :returns: ``self.plugins`` """ for plugin_path in self.find_plugins(): try: plugin = self.load_plugin(plugin_path) if plugin and plugin.name not in self.plugins: self.plugins[plugin.name] = plugin log.info('%s, loaded plugin: %s', self, plugin.name) # NOTE: prevent silent, implicit overwrite here (two plugins in two diff directories) # TODO: overwriting may be desired elif plugin and plugin.name in self.plugins: log.warning('%s, plugin with name already exists: %s. Skipping...', self, plugin.name) except Exception: if not self.skip_bad_plugins: raise log.exception('Plugin loading raised exception: %s. Skipping...', plugin_path) return self.plugins
[docs] def find_plugins(self): """ Return the directory paths of plugins within ``self.directories``. Paths are considered a plugin path if they pass ``self.is_plugin``. :rtype: string generator :returns: paths of valid plugins """ # due to the ordering of listdir, there is an implicit plugin loading order here # could instead explicitly list on/off in master config file for directory in self.directories: for plugin_dir in sorted(os.listdir(directory)): plugin_path = os.path.join(directory, plugin_dir) if self.is_plugin(plugin_path): yield plugin_path
[docs] def is_plugin(self, plugin_path): """ Determines whether the given filesystem path contains a plugin. In this base class, all sub-directories are considered plugins. :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the potential plugin :rtype: bool :returns: True if the path contains a plugin """ if not os.path.isdir(plugin_path): return False return True
[docs] def load_plugin(self, plugin_path): """ Create, load, and/or initialize the plugin and return it. Plugin bunches are decorated with: * name : the plugin name * path : the plugin path :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the plugin :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ plugin = bunch.Bunch( # TODO: need a better way to define plugin names # pro: filesystem name ensures uniqueness # con: rel. inflexible name=os.path.split(plugin_path)[1], path=plugin_path ) return plugin
# ============================================================================= plugin managers using hooks
[docs]class HookPluginManager(PluginManager): """ A hook plugin is a directory containing python modules or packages that: * allow creating, including, and running custom code at specific 'hook' points/events * are not tracked in the Galaxy repository and allow adding custom code to a Galaxy installation A HookPluginManager imports the plugin code needed and calls the plugin's hook functions at the specified time. """ #: the python file that will be imported - hook functions should be contained here loading_point_filename = 'plugin.py' hook_fn_prefix = 'hook_'
[docs] def is_plugin(self, plugin_path): """ Determines whether the given filesystem path contains a hookable plugin. All sub-directories that contain ``loading_point_filename`` are considered plugins. :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the potential plugin :rtype: bool :returns: True if the path contains a plugin """ if not super(HookPluginManager, self).is_plugin(plugin_path): return False # TODO: possibly switch to <plugin.name>.py or __init__.py if self.loading_point_filename not in os.listdir(plugin_path): return False return True
[docs] def load_plugin(self, plugin_path): """ Import the plugin ``loading_point_filename`` and attach to the plugin bunch. Plugin bunches are decorated with: * name : the plugin name * path : the plugin path * module : the plugin code :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the plugin :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ plugin = super(HookPluginManager, self).load_plugin(plugin_path) loading_point_name = self.loading_point_filename[:-3] plugin['module'] = self.import_plugin_module(loading_point_name, plugin) return plugin
[docs] def import_plugin_module(self, loading_point_name, plugin, import_as=None): """ Import the plugin code and cache the module in the plugin object. :type loading_point_name: string :param loading_point_name: name of the python file to import (w/o extension) :type plugin: ``util.bunch.Bunch`` :param plugin: the plugin containing the template to render :type import_as: string :param import_as: namespace to use for imported module This will be prepended with the ``__name__`` of this file. Defaults to ``plugin.name`` :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ # add this name to import_as (w/ default to plugin.name) to prevent namespace pollution in sys.modules import_as = '%s.%s' % (__name__, (import_as or plugin.name)) module_file, pathname, description = imp.find_module(loading_point_name, [plugin.path]) try: # TODO: hate this hack but only way to get package imports inside the plugin to work? sys.path.append(plugin.path) # sys.modules will now have import_as in its list module = imp.load_module(import_as, module_file, pathname, description) finally: module_file.close() if plugin.path in sys.path: sys.path.remove(plugin.path) return module
[docs] def run_hook(self, hook_name, *args, **kwargs): """ Search all plugins for a function named ``hook_fn_prefix`` + ``hook_name`` and run it passing in args and kwargs. Return values from each hook are returned in a dictionary keyed with the plugin names. :type hook_name: string :param hook_name: name (suffix) of the hook to run :rtype: dictionary :returns: where keys are plugin.names and values return values from the hooks """ # TODO: is hook prefix necessary? # TODO: could be made more efficient if cached by hook_name in the manager on load_plugin # (low maint. overhead since no dynamic loading/unloading of plugins) hook_fn_name = ''.join([self.hook_fn_prefix, hook_name]) returned = {} for plugin_name, plugin in self.plugins.items(): hook_fn = getattr(plugin.module, hook_fn_name, None) if hook_fn and hasattr(hook_fn, '__call__'): try: fn_returned = hook_fn(*args, **kwargs) returned[plugin.name] = fn_returned except Exception: # fail gracefully and continue with other plugins log.exception('Hook function "%s" failed for plugin "%s"', hook_name, plugin.name) # not sure of utility of this - seems better to be fire-and-forget pub-sub return returned
[docs] def filter_hook(self, hook_name, hook_arg, *args, **kwargs): """ Search all plugins for a function named ``hook_fn_prefix`` + ``hook_name`` and run the first with ``hook_arg`` and every function after with the return value of the previous. ..note: This makes plugin load order very important. :type hook_name: string :param hook_name: name (suffix) of the hook to run :type hook_arg: any :param hook_arg: the arg to be passed between hook functions :rtype: any :returns: the modified hook_arg """ hook_fn_name = ''.join([self.hook_fn_prefix, hook_name]) for plugin_name, plugin in self.plugins.items(): hook_fn = getattr(plugin.module, hook_fn_name, None) if hook_fn and hasattr(hook_fn, '__call__'): try: hook_arg = hook_fn(hook_arg, *args, **kwargs) except Exception: # fail gracefully and continue with other plugins log.exception('Filter hook function "%s" failed for plugin "%s"', hook_name, plugin.name) # may have been altered by hook fns, return return hook_arg
[docs]class PluginManagerStaticException(PluginManagerException): """Exception for plugin framework static directory set up errors. """ pass
[docs]class PluginManagerTemplateException(PluginManagerException): """Exception for plugin framework template directory and template rendering errors. """ pass
# ============================================================================= base
[docs]class PageServingPluginManager(PluginManager): """ Page serving plugins are files/directories that: * are not tracked in the Galaxy repository and allow adding custom code to a Galaxy installation * serve static files (css, js, images, etc.), * render templates A PageServingPluginManager sets up all the above components. """ # TODO: I'm unclear of the utility of this class - it prob. will only have one subclass (vis reg). Fold into? #: default static url base DEFAULT_BASE_URL = '' #: does the class need static files served? serves_static = True #: does the class need template files served? serves_templates = True #: default number of templates to search for plugin template lookup DEFAULT_TEMPLATE_COLLECTION_SIZE = 10 #: default encoding of plugin templates DEFAULT_TEMPLATE_ENCODING = 'utf-8' #: name of files to search for additional template lookup directories additional_template_paths_config_filename = 'additional_template_paths.xml'
[docs] def __init__(self, app, base_url='', template_cache_dir=None, **kwargs): """ Set up the manager and load all plugins. :type app: UniverseApplication :param app: the application (and its configuration) using this manager :type base_url: string :param base_url: url to prefix all plugin urls with :type template_cache_dir: string :param template_cache_dir: filesytem path to the directory where cached templates are kept """ self.base_url = base_url or self.DEFAULT_BASE_URL if not self.base_url: raise PluginManagerException('base_url or DEFAULT_BASE_URL required') self.template_cache_dir = template_cache_dir self.additional_template_paths = [] super(PageServingPluginManager, self).__init__(app, **kwargs)
[docs] def load_configuration(self): """ Load framework wide configuration, including: additional template lookup directories """ for directory in self.directories: possible_path = os.path.join(directory, self.additional_template_paths_config_filename) if os.path.exists(possible_path): added_paths = self.parse_additional_template_paths(possible_path, directory) self.additional_template_paths.extend(added_paths)
[docs] def parse_additional_template_paths(self, config_filepath, base_directory): """ Parse an XML config file at `config_filepath` for template paths (relative to `base_directory`) to add to each plugin's template lookup. Allows having a set of common templates for import/inheritance in plugin templates. :type config_filepath: string :param config_filepath: filesystem path to the config file :type base_directory: string :param base_directory: path prefixed to new, relative template paths """ additional_paths = [] xml_tree = util.parse_xml(config_filepath) paths_list = xml_tree.getroot() for rel_path_elem in paths_list.findall('path'): if rel_path_elem.text is not None: additional_paths.append(os.path.join(base_directory, rel_path_elem.text)) return additional_paths
[docs] def is_plugin(self, plugin_path): """ Determines whether the given filesystem path contains a plugin. If the manager ``serves_templates`` and a sub-directory contains another sub-directory named 'templates' it's considered valid. If the manager ``serves_static`` and a sub-directory contains another sub-directory named 'static' it's considered valid. :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the potential plugin :rtype: bool :returns: True if the path contains a plugin """ if not super(PageServingPluginManager, self).is_plugin(plugin_path): return False # reject only if we don't have either listdir = os.listdir(plugin_path) if(('templates' not in listdir) and ('static' not in listdir)): return False return True
[docs] def load_plugin(self, plugin_path): """ Create the plugin and decorate with static and/or template paths and urls. Plugin bunches are decorated with: * name : the plugin name * path : the plugin path * base_url : a url to the plugin :type plugin_path: string :param plugin_path: relative or absolute filesystem path to the plugin :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ plugin = super(PageServingPluginManager, self).load_plugin(plugin_path) # TODO: urlencode? plugin['base_url'] = '/'.join([self.base_url, plugin.name]) plugin = self._set_up_static_plugin(plugin) plugin = self._set_up_template_plugin(plugin) return plugin
def _set_up_static_plugin(self, plugin): """ Decorate the plugin with paths and urls needed to serve static content. Plugin bunches are decorated with: * serves_static : whether this plugin will serve static content If the plugin path contains a 'static' sub-dir, the following are added: * static_path : the filesystem path to the static content * static_url : the url to use when serving static content :type plugin: ``util.bunch.Bunch`` :param plugin: the plugin to decorate :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ plugin['serves_static'] = False static_path = os.path.join(plugin.path, 'static') if self.serves_static and os.path.isdir(static_path): plugin.serves_static = True plugin['static_path'] = static_path plugin['static_url'] = '/'.join([plugin.base_url, 'static']) return plugin def _set_up_template_plugin(self, plugin): """ Decorate the plugin with paths needed to fill templates. Plugin bunches are decorated with: * serves_templates : whether this plugin will use templates If the plugin path contains a 'static' sub-dir, the following are added: * template_path : the filesystem path to the template sub-dir * template_lookup : the (currently Mako) TemplateLookup used to search for templates :type plugin: ``util.bunch.Bunch`` :param plugin: the plugin to decorate :rtype: ``util.bunch.Bunch`` :returns: the loaded plugin object """ plugin['serves_templates'] = False template_path = os.path.join(plugin.path, 'templates') if self.serves_templates and os.path.isdir(template_path): plugin.serves_templates = True plugin['template_path'] = template_path plugin['template_lookup'] = self.build_plugin_template_lookup(plugin) return plugin # ------------------------------------------------------------------------- serving static files
[docs] def get_static_urls_and_paths(self): """ For each plugin, return a 2-tuple where the first element is a url path to the plugin's static files and the second is a filesystem path to those same files. Meant to be passed to a Static url map. :rtype: list of 2-tuples :returns: all urls and paths for each plugin serving static content """ # called during the static middleware creation (buildapp.py, wrap_in_static) urls_and_paths = [] for plugin in self.plugins.values(): if plugin.serves_static: urls_and_paths.append((plugin.static_url, plugin.static_path)) return urls_and_paths
# ------------------------------------------------------------------------- templates
[docs] def build_plugin_template_lookup(self, plugin): """ Builds the object that searches for templates (cached or not) when rendering. :type plugin: ``util.bunch.Bunch`` :param plugin: the plugin containing the templates :rtype: ``Mako.lookup.TemplateLookup`` :returns: template lookup for this plugin """ if not plugin.serves_templates: return None template_lookup_paths = plugin.template_path if self.additional_template_paths: template_lookup_paths = [template_lookup_paths] + self.additional_template_paths template_lookup = self._create_mako_template_lookup(self.template_cache_dir, template_lookup_paths) return template_lookup
def _create_mako_template_lookup(self, cache_dir, paths, collection_size=DEFAULT_TEMPLATE_COLLECTION_SIZE, output_encoding=DEFAULT_TEMPLATE_ENCODING): """ Create a ``TemplateLookup`` with defaults. :rtype: ``Mako.lookup.TemplateLookup`` :returns: all urls and paths for each plugin serving static content """ # TODO: possible to add galaxy/templates into the lookup here? return mako.lookup.TemplateLookup( directories=paths, module_directory=cache_dir, collection_size=collection_size, output_encoding=output_encoding)
[docs] def fill_template(self, trans, plugin, template_filename, **kwargs): """ Pass control over to trans and render ``template_filename``. :type trans: ``galaxy.web.framework.webapp.GalaxyWebTransaction`` :param trans: transaction doing the rendering :type plugin: ``util.bunch.Bunch`` :param plugin: the plugin containing the template to render :type template_filename: string :param template_filename: the path of the template to render relative to ``plugin.template_path`` :returns: rendered template """ # defined here to be overridden return trans.fill_template(template_filename, template_lookup=plugin.template_lookup, **kwargs)
# TODO: add fill_template fn that is able to load extra libraries beforehand (and remove after) # TODO: add template helpers specific to the plugins # TODO: some sort of url_for for these plugins # =============================================================================
[docs]class Plugin(object): """ Plugin as object/class. """
[docs] def __init__(self, app, path, name, config, context=None, **kwargs): context = context or {} self.app = app self.path = path self.name = name self.config = config