view env/lib/python3.9/site-packages/planemo/context.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

"""Define the context around Planemo computation.

Abstractions for cross cutting concerns (logging, workspace management,
etc.).
"""
import abc
import logging.config
import os
import shutil
import sys
import traceback
from urllib.request import urlopen

from planemo.config import read_global_config


class PlanemoContextInterface(metaclass=abc.ABCMeta):
    """Interface Planemo operations use to access workspace context."""

    @abc.abstractmethod
    def set_option_source(self, param_name, option_source, force=False):
        """Specify how an option was set."""

    @abc.abstractmethod
    def get_option_source(self, param_name):
        """Return OptionSource value indicating how the option was set."""

    @abc.abstractproperty
    def global_config(self):
        """Read Planemo's global configuration."""

    @abc.abstractmethod
    def log(self, msg, *args):
        """Log a message."""

    @abc.abstractmethod
    def vlog(self, msg, *args, **kwds):
        """Log a message only if verbose is enabled."""

    @abc.abstractproperty
    def workspace(self):
        """Create and return Planemo's workspace."""

    @abc.abstractproperty
    def galaxy_profiles_directory(self):
        """Create a return a directory for storing Galaxy profiles."""

    @abc.abstractmethod
    def cache_download(self, url, destination):
        """Use workspace to cache download of ``url``."""


class PlanemoContext(PlanemoContextInterface):
    """Implementation of ``PlanemoContextInterface``"""

    def __init__(self):
        """Construct a Context object using execution environment."""
        self.home = os.getcwd()
        self._global_config = None
        # Will be set by planemo CLI driver
        self.verbose = False
        self.planemo_config = None
        self.planemo_directory = None
        self.option_source = {}

    def set_option_source(self, param_name, option_source, force=False):
        """Specify how an option was set."""
        if not force:
            assert param_name not in self.option_source, "No option source for [%s]" % param_name
        self.option_source[param_name] = option_source

    def get_option_source(self, param_name):
        """Return OptionSource value indicating how the option was set."""
        assert param_name in self.option_source, "No option source for [%s]" % param_name
        return self.option_source[param_name]

    @property
    def global_config(self):
        """Read Planemo's global configuration.

        As defined most simply by ~/.planemo.yml.
        """
        if self._global_config is None:
            self._global_config = read_global_config(self.planemo_config)
        return self._global_config or {}

    def log(self, msg, *args):
        """Log a message."""
        if args:
            msg %= args
        self._log_message(msg)

    def vlog(self, msg, *args, **kwds):
        """Log a message only if verbose is enabled."""
        if self.verbose:
            self.log(msg, *args)
            if kwds.get("exception", False):
                traceback.print_exc(file=sys.stderr)

    @property
    def workspace(self):
        """Create and return Planemo's workspace.

        By default this will be ``~/.planemo``.
        """
        if not self.planemo_directory:
            raise Exception("No planemo workspace defined.")
        workspace = self.planemo_directory
        return self._ensure_directory(workspace, "workspace")

    @property
    def galaxy_profiles_directory(self):
        """Create a return a directory for storing Galaxy profiles."""
        path = os.path.join(self.workspace, "profiles")
        return self._ensure_directory(path, "Galaxy profiles")

    def cache_download(self, url, destination):
        """Use workspace to cache download of ``url``."""
        cache = os.path.join(self.workspace, "cache")
        if not os.path.exists(cache):
            os.makedirs(cache)
        filename = os.path.basename(url)
        cache_destination = os.path.join(cache, filename)
        if not os.path.exists(cache_destination):
            with urlopen(url) as fh:
                content = fh.read()
            if len(content) == 0:
                raise Exception("Failed to download [%s]." % url)
            with open(cache_destination, "wb") as f:
                f.write(content)

        shutil.copy(cache_destination, destination)

    def _ensure_directory(self, path, name):
        if not os.path.exists(path):
            os.makedirs(path)
        if not os.path.isdir(path):
            template = "Planemo %s directory [%s] unavailable."
            message = template % (name, path)
            raise Exception(message)
        return path

    def _log_message(self, message):
        """Extension point for overriding I/O."""
        print(message)


def configure_standard_planemo_logging(verbose):
    """Configure Planemo's default logging rules."""
    logging_config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'verbose': {
                'format': '%(name)s %(levelname)s %(asctime)s: %(message)s'
            },
            'simple': {
                'format': '%(name)s %(levelname)s: %(message)s'
            },
        },
        'handlers': {
            'console': {
                'level': 'DEBUG',
                'class': 'logging.StreamHandler',
                'formatter': 'simple' if not verbose else 'verbose'
            },
        },
        'loggers': {
            # Suppress CWL is beta warning, for Planemo purposes - it is absolutely not.
            'galaxy.tools.parser.factory': {
                'handlers': ['console'],
                'propagate': False,
                'level': 'ERROR' if not verbose else "DEBUG",
            },
            'galaxy.tools.deps.commands': {
                'handlers': ['console'],
                'propagate': False,
                'level': 'ERROR' if not verbose else "DEBUG",
            },
            'galaxy': {
                'handlers': ['console'],
                'propagate': False,
                'level': 'INFO' if not verbose else "DEBUG",
            },
            # @jmchilton
            # I'm fixing up Planemo's lint functionality for CWL and I keep seeing this for the
            # schema metadata stuff (e.g. in the workflows repo). "rdflib.term WARNING:
            # http://schema.org/docs/!DOCTYPE html does not look like a valid URI, trying to
            # serialize this will break.". I'm going to suppress this warning I think, or are the
            # examples wrong and should declare their namespaces differently in some way?
            # @mr-c
            # That particular warning is worth suppressing. A PR to silence it permanently would be very welcome!
            # https://github.com/RDFLib/rdflib/blob/master/rdflib/term.py#L225
            'rdflib.term': {
                'handlers': ['console'],
                'propagate': False,
                'level': 'ERROR' if not verbose else "DEBUG",
            }
        },
        'root': {
            'handlers': ['console'],
            'propagate': False,
            'level': 'WARNING' if not verbose else "DEBUG",
        }
    }
    logging.config.dictConfig(logging_config)


__all__ = (
    'configure_standard_planemo_logging',
    'PlanemoContextInterface',
    'PlanemoContext',
)