Source code for thoth.analyzer.command

#!/usr/bin/env python3
# thoth-analyzer
# Copyright(C) 2018, 2019, 2020 Fridolin Pokorny
# This program is free software: you can redistribute it and / or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <>.

"""Handling invoking commands of external programs in a sane way."""

import json
import logging

import delegator

_LOG = logging.getLogger(__name__)

[docs]class CommandResult(object): """Representation of result of a command invocation.""" def __init__(self, command: delegator.Command, is_json: bool = False): """Initialiaztion of a command result wrapper for delegator.""" self.command = command self.is_json = is_json self._stdout = None @property def stdout(self): """Return standard output from invocation.""" if self._stdout is None: if self.is_json: self._stdout = json.loads(self.command.out) else: self._stdout = self.command.out return self._stdout @property def stderr(self): """Return standard error output from invocation.""" return self.command.err @property def return_code(self): """Process return code.""" return self.command.return_code @property def timeout(self): """Timeout that was given to the invoked process to finish.""" return self.command.timeout
[docs] def to_dict(self): """Represent command result as a dict.""" return { "stdout": self.stdout, "stderr": self.stderr, "return_code": self.return_code, "command": self.command.cmd, "timeout": self.timeout, }
[docs]class CommandError(RuntimeError, CommandResult): """Exception raised on error when calling commands. Note that this class inherits also from CommandResult, so you can directly access to_dict() or other defined methods. """ def __init__(self, *args, command: delegator.Command, **command_result_kwargs): """Store information about command error.""" RuntimeError.__init__(self, *args) CommandResult.__init__(self, command=command, **command_result_kwargs) @property def stdout(self): """Return standard output from invocation. Override implementation for errors, not all tools product JSON or errors so try to avoid parsing JSON implicitly. """ return self.command.out
[docs]def run_command(cmd, timeout=60, is_json=False, env=None, raise_on_error=True): """Run the given command, block until it finishes.""" _LOG.debug("Running command %r", cmd) command =, block=True, timeout=timeout, env=env) if command.return_code != 0 and raise_on_error: error_msg = "Command exited with non-zero status code ({}): {}".format( command.return_code, command.err ) _LOG.debug(error_msg) raise CommandError(error_msg, command=command, is_json=is_json) return CommandResult(command, is_json=is_json)