codekingpro/portable-devtools

1from __future__ import annotations2 3 4__lazy_modules__ = ['subprocess']5 6import contextvars7import logging8import subprocess9import typing10 11 12TYPE_CHECKING = False13 14if TYPE_CHECKING:15    from collections.abc import Mapping, Sequence16 17    from ._types import StrPath18 19 20class Logger(typing.Protocol):  # pragma: no cover21    def __call__(self, message: str, *, kind: tuple[str, ...] | None = None) -> None: ...22 23 24_package_name = __spec__.parent25_default_logger = logging.getLogger(_package_name)26 27 28def _log_default(message: str, *, kind: tuple[str, ...] | None = None) -> None:  # noqa: ARG00129    # the log function that works in tests, real log function is set in __main__30    _default_logger.log(logging.INFO, message, stacklevel=2)31 32 33LOGGER = contextvars.ContextVar('LOGGER', default=_log_default)34VERBOSITY = contextvars.ContextVar('VERBOSITY', default=0)35 36 37def log_subprocess_error(error: subprocess.CalledProcessError) -> None:38    log = LOGGER.get()39 40    log(subprocess.list2cmdline(error.cmd), kind=('subprocess', 'cmd'))41 42    for stream_name in ('stdout', 'stderr'):43        stream = getattr(error, stream_name)44        if stream:45            log(stream.decode() if isinstance(stream, bytes) else stream, kind=('subprocess', stream_name))46 47 48def run_subprocess(cmd: Sequence[StrPath], cwd: str | None = None, env: Mapping[str, str] | None = None) -> None:49    verbosity = VERBOSITY.get()50 51    if verbosity > 0:52        import concurrent.futures53 54        log = LOGGER.get()55 56        with (57            concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor,58            subprocess.Popen(  # noqa: S60359                cmd, cwd=cwd, encoding='utf-8', env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT60            ) as process,61        ):62            log(subprocess.list2cmdline(cmd), kind=('subprocess', 'cmd'))63 64            @executor.submit65            def log_stream() -> None:66                assert process.stdout67                for line in process.stdout:68                    log(line, kind=('subprocess', 'stdout'))69 70            concurrent.futures.wait([log_stream])71 72            code = process.wait()73            if code:  # pragma: no cover74                raise subprocess.CalledProcessError(code, process.args)75 76    else:77        try:78            subprocess.run(cmd, capture_output=True, check=True, cwd=cwd, env=env)  # noqa: S60379        except subprocess.CalledProcessError as error:80            log_subprocess_error(error)81            raise82 83 84if TYPE_CHECKING:85    log: Logger86    verbosity: bool87 88else:89 90    def __getattr__(name: str) -> object:91        if name == 'log':92            return LOGGER.get()93        if name == 'verbosity':94            return VERBOSITY.get()95        raise AttributeError(name)  # pragma: no cover96 97 98__all__ = [99    'LOGGER',100    'VERBOSITY',101    'log',102    'log_subprocess_error',103    'run_subprocess',104    'verbosity',105]106