jw-pkg/src/python/jw/pkg/lib/ExecContext.py

# -*- 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):
        self.__interactive = interactive

    @property
    def interactive(self):
        return self.__interactive

    @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.
        """
        return await self._run(
            args=args,
            wd=wd,
            throw=throw,
            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=True) -> Result:
        pass

    async def sudo(self, cmd: list[str], mod_env: dict[str, str] = {}, opts: list[str]=[], verbose=True) -> Result:
        return await self._sudo(cmd, mod_env, opts, verbose)