Source code for polyvers.utils.oscmd

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 2015-2018 European Commission (JRC);
# Licensed under the EUPL 1.2+ (the 'Licence');
# You may not use this work except in compliance with the Licence.
# You may obtain a copy of the Licence at:
"""Utility to call OS commands through :func:`` with logging.

The *polyvers* version-configuration tool is generating tags like::


On purpose python code here kept with as few dependencies as possible."""

from typing import Dict
import logging

import subprocess as sbp

#: Monkeypatch :class:`subprocess.CalledProcessError`
#: to always print STDERR on errors.
#: TODO: reuse from `polyversion, include CWD.
def err_includes_stderr(self):
    import signal

    tail = ('\n  STDERR: %s' % self.stderr) if self.stderr else ''
    tail += ('\n  STDOUT: %s' % self.stdout) if self.stdout else ''

    if self.returncode and self.returncode < 0:
            return "Command '%s' died with %r.%s" % (
                self.cmd, signal.Signals(-self.returncode), tail)
        except ValueError:
            return "Command '%s' died with unknown signal %d.%s" % (
                self.cmd, -self.returncode, tail)
        return "Command '%s' returned non-zero exit status %d.%s" % (
            self.cmd, self.returncode, tail)

sbp.CalledProcessError.__str__ = err_includes_stderr

def format_syscmd(cmd):
    if isinstance(cmd, (list, tuple)):
        cmd = ' '.join('"%s"' % s if ' ' in s else s
                       for s in cmd)
        assert isinstance(cmd, str), cmd

    return cmd

class _CmdName:
    "To `workaround `:classLFileNotFoundError` not explaing it's a command missing."
    def __init__(self, path):
        self.path = path

    def __repr__(self):
        return "'%s' (command)" % self.path

[docs]def exec_cmd(cmd, dry_run=False, check_stdout=True, check_stderr=True, check_returncode=True, encoding='utf-8', encoding_errors='surrogateescape', **popen_kws): """ :param check_stdout: None: Popen(stdout=None), printed False: Popen(stdout=sbp.DEVNULL), ignored True: Popen(stdout=sbp.PIPE), collected & returned """ log = logging.getLogger(__name__) call_types = { None: {'label': 'EXEC', 'stream': None}, False: {'label': 'EXEC(no-stdout)', 'stream': sbp.DEVNULL}, True: {'label': 'CALL', 'stream': sbp.PIPE}, } stdout_ctype = call_types[check_stdout] cmd_label = stdout_ctype['label'] cmd_str = format_syscmd(cmd) log.debug('%s%s %r', 'DRY_' if dry_run else '', cmd_label, cmd_str) if dry_run: return try: ##WARN: python 3.6 `encoding` & `errors` kwds in `Popen`. res: sbp.CompletedProcess = cmd, stdout=stdout_ctype['stream'], stderr=call_types[check_stderr]['stream'], encoding=encoding, errors=encoding_errors, **popen_kws ) except FileNotFoundError as ex: ## On Windows you don't see the command attempted to run: # FileNotFoundError: [WinError 2] The system cannot find the file specified # if not ex.filename: ex.filename = _CmdName(cmd[0]) raise if res.returncode: log.log( logging.DEBUG if check_returncode else logging.WARNING, '%s %r failed with %s!\n stdout: %s\n stderr: %s', cmd_label, cmd_str, res.returncode, res.stdout, res.stderr) elif check_stdout or check_stderr: log.debug('%s %r ok: \n stdout: %s\n stderr: %s', cmd_label, cmd_str, res.stdout, res.stderr) if check_returncode: try: res.check_returncode() except sbp.CalledProcessError as ex: # # A hackish wedge to report to the user ## the CWD when not in Git-repo. if "ot a git repository" in (ex.stderr.decode() if isinstance(ex.stderr, bytes) else ex.stderr): import os from .. import pvtags raise pvtags.NoGitRepoError( "Current-dir '%s' is not within a git repository!" % os.curdir) from ex raise return res
def _as_flag(k): return k.replace('_', '-') class _Cli: def __init__(self, popen_kw: Dict, cmd: str): self._popen_kw = popen_kw self._cmdlist = [cmd] def _extend_cmdlist(self, args, kw): def kv2args(k, v): nk = len(k) if nk > 1: k = _as_flag(k) if v is None: return () if isinstance(v, bool): if v: flag = '-' + k if len(k) == 1 else '--' + k else: if nk == 1: raise ValueError( "Cannot negate single-letter flag '-%s'!" "\n cmd: %s!" % (k, ' '.join(self._cmdlist))) flag = '--no-' + k return (flag, ) return ('-%s' % k, str(v)) if nk == 1 else ('--%s=%s' % (k, v), ) arglist = self._cmdlist arglist.extend(str(a) for a in args) arglist.extend(arg for kv in kw.items() for arg in kv2args(*kv)) def __getattr__(self, attr): if attr == '__wrapped__': # PYTEST MAGIC! return None if attr: attr = _as_flag(attr) self._cmdlist.append(attr) return self def __call__(self, *args, **kw) -> str: self._extend_cmdlist(args, kw) res = exec_cmd(self._cmdlist, **self._popen_kw) self.rc = res.returncode self.stderr = res.stderr self.stdout = res.stdout # keep unstripped stdout if self._popen_kw['check_stdout']: return res.stdout and res.stdout.strip() def _(self, *args, **kw): "Avoid immediate execution and continue building cmd-line." self._extend_cmdlist(args, kw) return self def __str__(self): return 'Cli(%s)' % ' '.join("''" if a == '' else a for a in self._cmdlist)
[docs]class PopenCmd: """ A function --> cmd-line builder for executing (mostly) git commands. To run ``git log -n1``:: out = cmd.git.log(n=1) To launch a short python program with ``python -c "print('a')"``:: out = cmd.python._(c=True)('print(\'a\')') :raise sbp.CalledProcessError: if check_returncode=true and exit code is non-zero. .. Important:: Flags written out are mostly for Git, bc flags are produced like that:: -f <value> --flag=<value> """ def __init__(self, dry_run=False, check_stdout=True, check_stderr=True, check_returncode=True, **popen_kw): """ Set the Popen kw-args to use when the cmd will be executed. :param dry_run: log but don't actually exec cmd. :param check_stdout: None: Popen(stdout=None), printed False: Popen(stdout=sbp.DEVNULL), ignored True: Popen(stdout=sbp.PIPE), collected & returned :param check_stderr: same as `check_stdout` :param check_returncode: if true, raise `sbp.CalledProcessError` if return-code not 0. """ popen_kw.update(locals()) del popen_kw['self'], popen_kw['popen_kw'] self._popen_kw = popen_kw def __getattr__(self, attr): return _Cli(self._popen_kw, attr)
cmd = PopenCmd()