Source code for mafw.tools.shell_tools

#  Copyright 2025–2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""
Shell execution utilities.

This module provides unified functions for executing external commands via subprocess,
with support for dry-runs, environment merging, and standardized output.
"""

import os
import shlex
import subprocess
from pathlib import Path
from typing import Any

from rich.console import Console

CONSOLE = Console()
"""Shared console for standardized output."""


[docs] def _to_command_line(command: str | list[str]) -> str: """ Convert a command into a printable shell-like string. :param command: Command expressed as a string or tokenized list. :type command: str | list[str] :return: Human-readable command line. :rtype: str """ if isinstance(command, str): return command return shlex.join(command)
[docs] def run( command: str | list[str], *, cwd: Path | str | None = None, dry_run: bool = False, env: dict[str, str] | None = None, quiet: bool = False, **kwargs: Any, ) -> subprocess.CompletedProcess[Any]: """ Execute a subprocess command and return a completed process object. The command line is always printed with a standard prefix. In dry-run mode, execution is skipped and a successful synthetic ``CompletedProcess`` is returned. .. versionadded:: 2.2 Add the `quiet` parameter to do not display the command being executed :param command: Command to execute. :type command: str | list[str] :param cwd: Working directory for command execution. :type cwd: Path | str | None :param dry_run: If ``True``, print the command without executing it. :type dry_run: bool :param env: Optional environment variables to merge with the current environment. :type env: dict[str, str] | None :param quiet: Suppress the standard console banner when ``True``. :type quiet: bool :param kwargs: Additional keyword arguments forwarded to ``subprocess.run``. :type kwargs: Any :return: Completed process produced by the command execution. :rtype: subprocess.CompletedProcess[Any] """ command_str = _to_command_line(command) if not quiet: CONSOLE.print(f'🧩 Running: {command_str}') if dry_run: return subprocess.CompletedProcess(args=command, returncode=0, stdout='', stderr='') kwargs.setdefault('check', True) kwargs.setdefault('text', True) if env is not None: merged_env = os.environ.copy() merged_env.update(env) kwargs['env'] = merged_env if isinstance(command, str): kwargs.setdefault('shell', True) return subprocess.run(command, cwd=cwd, **kwargs)
[docs] def run_stdout(command: str | list[str], *, dry_run: bool = False, quiet: bool = False, **kwargs: Any) -> str: """ Execute a command and return stripped standard output. .. versionadded:: 2.2 Add the `quiet` parameter to do not display the command being executed :param command: Command to execute. :type command: str | list[str] :param dry_run: If ``True``, do not execute the command. :type dry_run: bool :param quiet: Suppress the standard console banner when ``True``. :type quiet: bool :param kwargs: Additional keyword arguments forwarded to ``run``. :type kwargs: Any :return: Standard output stripped of leading and trailing whitespace. :rtype: str """ kwargs['capture_output'] = True result = run(command, dry_run=dry_run, quiet=quiet, **kwargs) stdout = result.stdout if isinstance(stdout, bytes): return stdout.decode().strip() return (stdout or '').strip()