lib.Result: Add module
.lib.Result has grown enourmously in size and merits its own module.
For now, reexport it from .lib.base to not break all code containing
"from jw.lib.base import Result"
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
parent
00dff385bc
commit
de5be6b757
2 changed files with 156 additions and 153 deletions
154
src/python/jw/pkg/lib/Result.py
Normal file
154
src/python/jw/pkg/lib/Result.py
Normal file
|
|
@ -0,0 +1,154 @@
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
class Result:
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
stdout: bytes | None,
|
||||||
|
stderr: bytes | None,
|
||||||
|
status: int,
|
||||||
|
encoding: str = 'UTF-8',
|
||||||
|
strip: bool = True,
|
||||||
|
cmd: list[str] | None = None,
|
||||||
|
wd: str | None = None,
|
||||||
|
) -> None:
|
||||||
|
self.__stdout = stdout
|
||||||
|
self.__stderr = stderr
|
||||||
|
self.__status = status
|
||||||
|
self.__encoding = encoding
|
||||||
|
self.__strip = strip
|
||||||
|
self.__cmd = cmd
|
||||||
|
self.__wd = wd
|
||||||
|
|
||||||
|
def __decode(self, stdxxx: bytes | None) -> str | None:
|
||||||
|
if stdxxx is None:
|
||||||
|
return None
|
||||||
|
ret = stdxxx.decode(self.encoding)
|
||||||
|
if self.strip:
|
||||||
|
return ret.strip()
|
||||||
|
return ret
|
||||||
|
|
||||||
|
@property
|
||||||
|
def status(self) -> int | None:
|
||||||
|
return self.__status
|
||||||
|
|
||||||
|
@property
|
||||||
|
def encoding(self) -> str:
|
||||||
|
return self.__encoding
|
||||||
|
|
||||||
|
@encoding.setter
|
||||||
|
def encoding(self, value: str) -> None:
|
||||||
|
self.__encoding = value
|
||||||
|
|
||||||
|
def __stdout_footprint(self, quote = False) -> str:
|
||||||
|
if self.__stdout is None:
|
||||||
|
ret = ''
|
||||||
|
else:
|
||||||
|
ret = self.stdout_str[:20]
|
||||||
|
if quote:
|
||||||
|
ret = '"{ret}"'
|
||||||
|
return ret
|
||||||
|
|
||||||
|
def __repr__(self) -> str:
|
||||||
|
ret = f'{self.__status}:'
|
||||||
|
if self.status != 0:
|
||||||
|
ret += f' err: {self.stderr_str_or_none}'
|
||||||
|
else:
|
||||||
|
ret += f' out: {self.__stdout_footprint(quote=True)}'
|
||||||
|
return ret
|
||||||
|
|
||||||
|
@property
|
||||||
|
def strip(self) -> bool:
|
||||||
|
return self.__strip
|
||||||
|
|
||||||
|
@strip.setter
|
||||||
|
def strip(self, value: bool) -> None:
|
||||||
|
self.__strip = value
|
||||||
|
|
||||||
|
@property
|
||||||
|
def cmd(self) -> list[str] | None:
|
||||||
|
return self.__cmd
|
||||||
|
|
||||||
|
@cmd.setter
|
||||||
|
def cmd(self, value: list[str]) -> None:
|
||||||
|
self.__cmd = value
|
||||||
|
|
||||||
|
@property
|
||||||
|
def wd(self) -> str | None:
|
||||||
|
return self.__wd
|
||||||
|
|
||||||
|
@wd.setter
|
||||||
|
def wd(self, value: str) -> None:
|
||||||
|
self.__wd = value
|
||||||
|
|
||||||
|
def matches_error(self, pattern: str) -> bool:
|
||||||
|
if self.status == 0:
|
||||||
|
return False
|
||||||
|
err = self.stderr_str
|
||||||
|
if err is None:
|
||||||
|
return False
|
||||||
|
import re
|
||||||
|
return re.search(pattern, err) is not None
|
||||||
|
|
||||||
|
def __summarize(self, cmd: list[str] | None, wd: str | None = None) -> str:
|
||||||
|
from .util import pretty_cmd
|
||||||
|
if cmd is None:
|
||||||
|
cmd = self.__cmd
|
||||||
|
call = ''
|
||||||
|
if cmd is not None:
|
||||||
|
if wd is None:
|
||||||
|
wd = self.__wd
|
||||||
|
call = f'"{pretty_cmd(cmd, wd)}" '
|
||||||
|
ret = f'Command {call}has exited with status {self.__status}'
|
||||||
|
call = pretty_cmd(cmd, wd)
|
||||||
|
if self.status != 0:
|
||||||
|
ret += f' -> stderr="{self.__stderr!r}"'
|
||||||
|
else:
|
||||||
|
if self.__stdout:
|
||||||
|
ret += f' -> stdout has {len(self.__stdout)} bytes'
|
||||||
|
else:
|
||||||
|
ret += ' -> stdout = None'
|
||||||
|
return ret
|
||||||
|
|
||||||
|
def summarize(self, cmd: list[str] | None = None, wd: str | None = None) -> str:
|
||||||
|
return self.__summarize(cmd, wd)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def summary(self) -> str:
|
||||||
|
return self.__summarize(None, None)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stdout(self) -> bytes:
|
||||||
|
if self.__stdout is None:
|
||||||
|
raise Exception(f'Result has no standard output stream: {self.summary}')
|
||||||
|
return self.__stdout
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stdout_or_none(self) -> bytes | None:
|
||||||
|
return self.__stdout
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stdout_str_or_none(self) -> str | None:
|
||||||
|
return self.__decode(self.__stdout)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stdout_str(self) -> str:
|
||||||
|
return self.stdout.decode(self.__encoding)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stderr(self) -> bytes:
|
||||||
|
if self.__stderr is None:
|
||||||
|
raise Exception(f'Result has no standard error stream: {self.summary}')
|
||||||
|
return self.__stderr
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stderr_or_none(self) -> bytes | None:
|
||||||
|
return self.__stderr
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stderr_str_or_none(self) -> str | None:
|
||||||
|
return self.__decode(self.__stderr)
|
||||||
|
|
||||||
|
@property
|
||||||
|
def stderr_str(self) -> str:
|
||||||
|
return self.stderr.decode(self.__encoding)
|
||||||
|
|
@ -3,6 +3,8 @@ from __future__ import annotations
|
||||||
from enum import Enum, auto
|
from enum import Enum, auto
|
||||||
from typing import TYPE_CHECKING, NamedTuple, TypeAlias
|
from typing import TYPE_CHECKING, NamedTuple, TypeAlias
|
||||||
|
|
||||||
|
from .Result import Result as Result
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
import os
|
import os
|
||||||
|
|
||||||
|
|
@ -14,159 +16,6 @@ class InputMode(Enum):
|
||||||
|
|
||||||
Input: TypeAlias = InputMode | bytes | str
|
Input: TypeAlias = InputMode | bytes | str
|
||||||
|
|
||||||
class Result:
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
stdout: bytes | None,
|
|
||||||
stderr: bytes | None,
|
|
||||||
status: int,
|
|
||||||
encoding: str = 'UTF-8',
|
|
||||||
strip: bool = True,
|
|
||||||
cmd: list[str] | None = None,
|
|
||||||
wd: str | None = None,
|
|
||||||
) -> None:
|
|
||||||
self.__stdout = stdout
|
|
||||||
self.__stderr = stderr
|
|
||||||
self.__status = status
|
|
||||||
self.__encoding = encoding
|
|
||||||
self.__strip = strip
|
|
||||||
self.__cmd = cmd
|
|
||||||
self.__wd = wd
|
|
||||||
|
|
||||||
def __decode(self, stdxxx: bytes | None) -> str | None:
|
|
||||||
if stdxxx is None:
|
|
||||||
return None
|
|
||||||
ret = stdxxx.decode(self.encoding)
|
|
||||||
if self.strip:
|
|
||||||
return ret.strip()
|
|
||||||
return ret
|
|
||||||
|
|
||||||
@property
|
|
||||||
def status(self) -> int | None:
|
|
||||||
return self.__status
|
|
||||||
|
|
||||||
@property
|
|
||||||
def encoding(self) -> str:
|
|
||||||
return self.__encoding
|
|
||||||
|
|
||||||
@encoding.setter
|
|
||||||
def encoding(self, value: str) -> None:
|
|
||||||
self.__encoding = value
|
|
||||||
|
|
||||||
def __stdout_footprint(self, quote = False) -> str:
|
|
||||||
if self.__stdout is None:
|
|
||||||
ret = ''
|
|
||||||
else:
|
|
||||||
ret = self.stdout_str[:20]
|
|
||||||
if quote:
|
|
||||||
ret = '"{ret}"'
|
|
||||||
return ret
|
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
|
||||||
ret = f'{self.__status}:'
|
|
||||||
if self.status != 0:
|
|
||||||
ret += f' err: {self.stderr_str_or_none}'
|
|
||||||
else:
|
|
||||||
ret += f' out: {self.__stdout_footprint(quote=True)}'
|
|
||||||
return ret
|
|
||||||
|
|
||||||
@property
|
|
||||||
def strip(self) -> bool:
|
|
||||||
return self.__strip
|
|
||||||
|
|
||||||
@strip.setter
|
|
||||||
def strip(self, value: bool) -> None:
|
|
||||||
self.__strip = value
|
|
||||||
|
|
||||||
@property
|
|
||||||
def cmd(self) -> list[str] | None:
|
|
||||||
return self.__cmd
|
|
||||||
|
|
||||||
@cmd.setter
|
|
||||||
def cmd(self, value: list[str]) -> None:
|
|
||||||
self.__cmd = value
|
|
||||||
|
|
||||||
@property
|
|
||||||
def wd(self) -> str | None:
|
|
||||||
return self.__wd
|
|
||||||
|
|
||||||
@wd.setter
|
|
||||||
def wd(self, value: str) -> None:
|
|
||||||
self.__wd = value
|
|
||||||
|
|
||||||
def matches_error(self, pattern: str) -> bool:
|
|
||||||
if self.status == 0:
|
|
||||||
return False
|
|
||||||
err = self.stderr_str
|
|
||||||
if err is None:
|
|
||||||
return False
|
|
||||||
import re
|
|
||||||
return re.search(pattern, err) is not None
|
|
||||||
|
|
||||||
def __summarize(self, cmd: list[str] | None, wd: str | None = None) -> str:
|
|
||||||
from .util import pretty_cmd
|
|
||||||
if cmd is None:
|
|
||||||
cmd = self.__cmd
|
|
||||||
call = ''
|
|
||||||
if cmd is not None:
|
|
||||||
if wd is None:
|
|
||||||
wd = self.__wd
|
|
||||||
call = f'"{pretty_cmd(cmd, wd)}" '
|
|
||||||
ret = f'Command {call}has exited with status {self.__status}'
|
|
||||||
call = pretty_cmd(cmd, wd)
|
|
||||||
if self.status != 0:
|
|
||||||
ret += f' -> stderr="{self.__stderr!r}"'
|
|
||||||
else:
|
|
||||||
if self.__stdout:
|
|
||||||
ret += f' -> stdout has {len(self.__stdout)} bytes'
|
|
||||||
else:
|
|
||||||
ret += ' -> stdout = None'
|
|
||||||
return ret
|
|
||||||
|
|
||||||
def summarize(self, cmd: list[str] | None = None, wd: str | None = None) -> str:
|
|
||||||
return self.__summarize(cmd, wd)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def summary(self) -> str:
|
|
||||||
return self.__summarize(None, None)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stdout(self) -> bytes:
|
|
||||||
if self.__stdout is None:
|
|
||||||
raise Exception(f'Result has no standard output stream: {self.summary}')
|
|
||||||
return self.__stdout
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stdout_or_none(self) -> bytes | None:
|
|
||||||
return self.__stdout
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stdout_str_or_none(self) -> str | None:
|
|
||||||
return self.__decode(self.__stdout)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stdout_str(self) -> str:
|
|
||||||
return self.stdout.decode(self.__encoding)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stderr(self) -> bytes:
|
|
||||||
if self.__stderr is None:
|
|
||||||
raise Exception(f'Result has no standard error stream: {self.summary}')
|
|
||||||
return self.__stderr
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stderr_or_none(self) -> bytes | None:
|
|
||||||
return self.__stderr
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stderr_str_or_none(self) -> str | None:
|
|
||||||
return self.__decode(self.__stderr)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def stderr_str(self) -> str:
|
|
||||||
return self.stderr.decode(self.__encoding)
|
|
||||||
|
|
||||||
class StatResult(NamedTuple):
|
class StatResult(NamedTuple):
|
||||||
mode: int
|
mode: int
|
||||||
owner: str
|
owner: str
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue