Source code for gwf.backends.logmanager

import io
import os.path

from ..utils import ensure_dir, redirect_exception
from .exceptions import LogError


class LogManager:
    """Base class for log managers.

    A log manager must as a minimum implement the :func:`open_stdout(target)`
    and :func:`open_stderr(target)` methods.

    Log managers are used by backends to specify how log files are stored and
    retrieved. Most backends will want to use the :class:`FileLogManager` which
    stores log files in the `.gwf/logs` directory. For testing of backends, the
    :class:`MemoryLogManager` can be useful.
    """

    def open_stdout(self, target, mode="r"):
        raise NotImplementedError()

    def open_stderr(self, target, mode="r"):
        raise NotImplementedError()

    def remove_stdout(self, target):
        raise NotImplementedError()

    def remove_stderr(self, target):
        raise NotImplementedError()

    def list(self):
        raise NotImplementedError()


[docs]class MemoryLogManager(LogManager): """A memory-based log manager. This log manager stores logs in memory. """ def __init__(self): self._stdout_logs = {} self._stderr_logs = {} @redirect_exception(KeyError, LogError) def open_stdout(self, target, mode="r"): if target not in self._stdout_logs and mode == "w": self._stdout_logs[target] = io.StringIO() return self._stdout_logs[target] @redirect_exception(KeyError, LogError) def open_stderr(self, target, mode="r"): if target not in self._stderr_logs and mode == "w": self._stderr_logs[target] = io.StringIO() return self._stderr_logs[target] @redirect_exception(KeyError, LogError) def remove_stdout(self, target): del self._stdout_logs[target] @redirect_exception(KeyError, LogError) def remove_stderr(self, target): del self._stderr_logs[target] def list(self): return set(self._stderr_logs.keys()).union(self._stdout_logs.keys())
[docs]class FileLogManager(LogManager): """A file-based log manager. This log manager stores logs on disk in the `log_dir` directory (which defaults to `.gwf/logs`). """ log_dir = ".gwf/logs" def __init__(self): super().__init__() ensure_dir(FileLogManager.log_dir) @staticmethod def _get_log_path(target_name, extension): """Return path for log file for a given target name. If `extension` is `stdout`, then the path of the file containing standard output of the target will be returned. If `stderr`, the path of the file containing standard error of the target will be returned. :arg target gwf.Target: Target to return log path for. :arg extension str: Must be either `stdout` or `stderr`. """ log_file = "{}.{}".format(target_name, extension) return os.path.join(FileLogManager.log_dir, log_file)
[docs] def stdout_path(self, target_name): """Return path of the log file containing standard output for target.""" return self._get_log_path(target_name, "stdout")
[docs] def stderr_path(self, target_name): """Return path of the log file containing standard error for target.""" return self._get_log_path(target_name, "stderr")
[docs] @redirect_exception(FileNotFoundError, LogError) def open_stdout(self, target, mode="r"): """Return file handle to the standard output log file for target. :raises LogError: If the log could not be found. """ return open(self.stdout_path(target.name), mode)
[docs] @redirect_exception(FileNotFoundError, LogError) def open_stderr(self, target, mode="r"): """Return file handle to standard error log file for target. :raises LogError: If the log could not be found. """ return open(self.stderr_path(target.name), mode)
@redirect_exception(OSError, LogError) def remove_stdout(self, target_name): os.remove(self.stdout_path(target_name)) @redirect_exception(OSError, LogError) def remove_stderr(self, target_name): os.remove(self.stderr_path(target_name)) def list(self): return ( os.path.splitext(log_name)[0] for log_name in os.listdir(FileLogManager.log_dir) )