mirror of
ssh://git.janware.com/janware/proj/jw-pkg
synced 2026-04-27 14:16:01 +02:00
Add the --verbose global option, which is made available as the App.verbose property. Some functions still take a verbose parameter, but the type of these parameters is converted from bool to bool|None. The idea is that, if they are None, their verbosity falls back to the global default. Signed-off-by: Jan Lindemann <jan@janware.com>
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import abc
|
|
from typing import NamedTuple
|
|
|
|
class Result(NamedTuple):
|
|
|
|
stdout: str|None
|
|
stderr: str|None
|
|
status: int|None
|
|
|
|
class ExecContext(abc.ABC):
|
|
|
|
def __init__(self, interactive: bool=True, verbose_default=False):
|
|
self.__interactive = interactive
|
|
self.__verbose_default = verbose_default
|
|
assert verbose_default is not None
|
|
|
|
def _verbose(self, verbose: bool|None) -> bool:
|
|
if verbose is not None:
|
|
return verbose
|
|
return self.__verbose_default
|
|
|
|
@property
|
|
def interactive(self) -> bool:
|
|
return self.__interactive
|
|
|
|
@property
|
|
def verbose_default(self) -> bool:
|
|
return self.__verbose_default
|
|
|
|
@abc.abstractmethod
|
|
async def _run(self, *args, **kwargs) -> Result:
|
|
pass
|
|
|
|
async def run(
|
|
self,
|
|
args: list[str],
|
|
wd: str|None = None,
|
|
throw: bool = True,
|
|
verbose: bool = False,
|
|
cmd_input: str|None = None,
|
|
env: dict[str, str]|None = None,
|
|
title: str=None,
|
|
output_encoding: str|None = None, # None => unchanged; "bytes" => return raw bytes
|
|
) -> Result:
|
|
"""
|
|
Run a command asynchronously and return its output
|
|
|
|
Args:
|
|
args: Command and arguments
|
|
wd: Optional working directory
|
|
throw: Raise an exception on non-zero exit status if True
|
|
verbose: Emit log output while the command runs
|
|
cmd_input:
|
|
- None -> stdin from /dev/null
|
|
- "mode:interactive" -> Inherit terminal stdin
|
|
- "mode:auto" -> Inherit terminal stdin if it is a TTY
|
|
- otherwise -> String fed to stdin
|
|
output_encoding:
|
|
- None -> unchanged behavior (decode stdout via sys.stdout.encoding, stderr via sys.stderr.encoding)
|
|
- "bytes" -> return raw bytes instead of decoded strings
|
|
- otherwise -> decode stdout/stderr using this encoding
|
|
|
|
Returns:
|
|
(stdout, stderr, exit_status):
|
|
stdout: stderr each as a string/bytes or None
|
|
In PTY mode stderr is always None because PTY merges stdout/stderr.
|
|
"""
|
|
|
|
if verbose is None:
|
|
verbose = self.__verbose_default
|
|
|
|
return await self._run(
|
|
args=args,
|
|
wd=wd,
|
|
throw=throw,
|
|
verbose=self._verbose(verbose),
|
|
cmd_input=cmd_input,
|
|
env=env,
|
|
title=title,
|
|
output_encoding=output_encoding
|
|
)
|
|
|
|
@abc.abstractmethod
|
|
async def _sudo(self, cmd: list[str], mod_env: dict[str, str], opts: list[str], verbose: bool) -> Result:
|
|
pass
|
|
|
|
async def sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose: bool|None=None) -> Result:
|
|
return await self._sudo(cmd, mod_env, opts, self._verbose(verbose))
|