zhangyiss/spack
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add type hints to spack.util.executable.Executable (#48468)

Browse Source 
* Add type-hints to `spack.util.executable.Executable`
* Add type-hint to input
* Use overload, and remove assertions at calling sites
* Bump mypy to v1.11.2 (working locally), Python to 3.13
This commit is contained in:
Massimiliano Culpo 2025-01-09 23:16:24 +01:00 committed by GitHub
parent 46da7952d3
commit 5085f635dd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 129 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/requirements/style/requirements.txt vendored
View File

@ -2,6 +2,6 @@ black==24.10.0
clingo==5.7.1
flake8==7.1.1
isort==5.13.2
mypy==1.8.0
mypy==1.11.2
types-six==1.17.0.20241205
vermin==1.6.0

6
.github/workflows/valid-style.yml vendored
View File

@ -20,7 +20,7 @@ jobs:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
- uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b
with:
python-version: '3.11'
python-version: '3.13'
cache: 'pip'
- name: Install Python Packages
run: |
@ -39,7 +39,7 @@ jobs:
fetch-depth: 0
- uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b
with:
python-version: '3.11'
python-version: '3.13'
cache: 'pip'
- name: Install Python packages
run: |
@ -58,7 +58,7 @@ jobs:
secrets: inherit
with:
with_coverage: ${{ inputs.with_coverage }}
python_version: '3.11'
python_version: '3.13'
# Check that spack can bootstrap the development environment on Python 3.6 - RHEL8
bootstrap-dev-rhel8:
runs-on: ubuntu-latest

214
lib/spack/spack/util/executable.py
View File

@ -7,11 +7,12 @@
import subprocess
import sys
from pathlib import Path, PurePath
from typing import Callable, Dict, Optional, Sequence, TextIO, Type, Union, overload
import llnl.util.tty as tty
import spack.error
import spack.util.environment
from spack.util.environment import EnvironmentModifications
__all__ = ["Executable", "which", "which_string", "ProcessError"]
@ -19,33 +20,29 @@
class Executable:
"""Class representing a program that can be run on the command line."""
def __init__(self, name):
def __init__(self, name: str) -> None:
file_path = str(Path(name))
if sys.platform != "win32" and name.startswith("."):
# pathlib strips the ./ from relative paths so it must be added back
file_path = os.path.join(".", file_path)
self.exe = [file_path]
self.default_env = {}
self.default_envmod = spack.util.environment.EnvironmentModifications()
self.returncode = None
self.default_env: Dict[str, str] = {}
self.default_envmod = EnvironmentModifications()
self.returncode = 0
self.ignore_quotes = False
if not self.exe:
raise ProcessError("Cannot construct executable for '%s'" % name)
def add_default_arg(self, *args):
def add_default_arg(self, *args: str) -> None:
"""Add default argument(s) to the command."""
self.exe.extend(args)
def with_default_args(self, *args):
def with_default_args(self, *args: str) -> "Executable":
"""Same as add_default_arg, but returns a copy of the executable."""
new = self.copy()
new.add_default_arg(*args)
return new
def copy(self):
def copy(self) -> "Executable":
"""Return a copy of this Executable."""
new = Executable(self.exe[0])
new.exe[:] = self.exe
@ -53,7 +50,7 @@ def copy(self):
new.default_envmod.extend(self.default_envmod)
return new
def add_default_env(self, key, value):
def add_default_env(self, key: str, value: str) -> None:
"""Set an environment variable when the command is run.
Parameters:
@ -62,68 +59,109 @@ def add_default_env(self, key, value):
"""
self.default_env[key] = value
def add_default_envmod(self, envmod):
def add_default_envmod(self, envmod: EnvironmentModifications) -> None:
"""Set an EnvironmentModifications to use when the command is run."""
self.default_envmod.extend(envmod)
@property
def command(self):
"""The command-line string.
Returns:
str: The executable and default arguments
"""
def command(self) -> str:
"""Returns the entire command-line string"""
return " ".join(self.exe)
@property
def name(self):
"""The executable name.
Returns:
str: The basename of the executable
"""
def name(self) -> str:
"""Returns the executable name"""
return PurePath(self.path).name
@property
def path(self):
"""The path to the executable.
Returns:
str: The path to the executable
"""
def path(self) -> str:
"""Returns the executable path"""
return str(PurePath(self.exe[0]))
def __call__(self, *args, **kwargs):
"""Run this executable in a subprocess.
@overload
def __call__(
self,
*args: str,
fail_on_error: bool = True,
ignore_errors: Union[int, Sequence[int]] = (),
ignore_quotes: Optional[bool] = None,
timeout: Optional[int] = None,
env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
input: Optional[TextIO] = None,
output: Union[Optional[TextIO], str] = None,
error: Union[Optional[TextIO], str] = None,
_dump_env: Optional[Dict[str, str]] = None,
) -> None: ...
@overload
def __call__(
self,
*args: str,
fail_on_error: bool = True,
ignore_errors: Union[int, Sequence[int]] = (),
ignore_quotes: Optional[bool] = None,
timeout: Optional[int] = None,
env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
input: Optional[TextIO] = None,
output: Union[Type[str], Callable] = ...,
error: Union[Optional[TextIO], str, Type[str], Callable] = None,
_dump_env: Optional[Dict[str, str]] = None,
) -> str: ...
@overload
def __call__(
self,
*args: str,
fail_on_error: bool = True,
ignore_errors: Union[int, Sequence[int]] = (),
ignore_quotes: Optional[bool] = None,
timeout: Optional[int] = None,
env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
input: Optional[TextIO] = None,
output: Union[Optional[TextIO], str, Type[str], Callable] = None,
error: Union[Type[str], Callable] = ...,
_dump_env: Optional[Dict[str, str]] = None,
) -> str: ...
def __call__(
self,
*args: str,
fail_on_error: bool = True,
ignore_errors: Union[int, Sequence[int]] = (),
ignore_quotes: Optional[bool] = None,
timeout: Optional[int] = None,
env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
extra_env: Optional[Union[Dict[str, str], EnvironmentModifications]] = None,
input: Optional[TextIO] = None,
output: Union[Optional[TextIO], str, Type[str], Callable] = None,
error: Union[Optional[TextIO], str, Type[str], Callable] = None,
_dump_env: Optional[Dict[str, str]] = None,
) -> Optional[str]:
"""Runs this executable in a subprocess.
Parameters:
*args (str): Command-line arguments to the executable to run
Keyword Arguments:
_dump_env (dict): Dict to be set to the environment actually
used (envisaged for testing purposes only)
env (dict or EnvironmentModifications): The environment with which
to run the executable
extra_env (dict or EnvironmentModifications): Extra items to add to
the environment (neither requires nor precludes env)
fail_on_error (bool): Raise an exception if the subprocess returns
an error. Default is True. The return code is available as
``exe.returncode``
ignore_errors (int or list): A list of error codes to ignore.
If these codes are returned, this process will not raise
an exception even if ``fail_on_error`` is set to ``True``
ignore_quotes (bool): If False, warn users that quotes are not needed
as Spack does not use a shell. Defaults to False.
timeout (int or float): The number of seconds to wait before killing
the child process
input: Where to read stdin from
output: Where to send stdout
error: Where to send stderr
*args: command-line arguments to the executable to run
fail_on_error: if True, raises an exception if the subprocess returns an error
The return code is available as ``self.returncode``
ignore_errors: a sequence of error codes to ignore. If these codes are returned, this
process will not raise an exception, even if ``fail_on_error`` is set to ``True``
ignore_quotes: if False, warn users that quotes are not needed, as Spack does not
use a shell. If None, use ``self.ignore_quotes``.
timeout: the number of seconds to wait before killing the child process
env: the environment with which to run the executable
extra_env: extra items to add to the environment (neither requires nor precludes env)
input: where to read stdin from
output: where to send stdout
error: where to send stderr
_dump_env: dict to be set to the environment actually used (envisaged for
testing purposes only)
Accepted values for input, output, and error:
* python streams, e.g. open Python file objects, or ``os.devnull``
* filenames, which will be automatically opened for writing
* ``str``, as in the Python string type. If you set these to ``str``,
output and error will be written to pipes and returned as a string.
If both ``output`` and ``error`` are set to ``str``, then one string
@ -133,8 +171,11 @@ def __call__(self, *args, **kwargs):
Behaves the same as ``str``, except that value is also written to
``stdout`` or ``stderr``.
By default, the subprocess inherits the parent's file descriptors.
For output and error it's also accepted:
* filenames, which will be automatically opened for writing
By default, the subprocess inherits the parent's file descriptors.
"""
def process_cmd_output(out, err):
@ -159,44 +200,34 @@ def process_cmd_output(out, err):
sys.stderr.write(errstr)
return result
# Environment
env_arg = kwargs.get("env", None)
# Setup default environment
env = os.environ.copy() if env_arg is None else {}
self.default_envmod.apply_modifications(env)
env.update(self.default_env)
current_environment = os.environ.copy() if env is None else {}
self.default_envmod.apply_modifications(current_environment)
current_environment.update(self.default_env)
# Apply env argument
if isinstance(env_arg, spack.util.environment.EnvironmentModifications):
env_arg.apply_modifications(env)
elif env_arg:
env.update(env_arg)
if isinstance(env, EnvironmentModifications):
env.apply_modifications(current_environment)
elif env:
current_environment.update(env)
# Apply extra env
extra_env = kwargs.get("extra_env", {})
if isinstance(extra_env, spack.util.environment.EnvironmentModifications):
extra_env.apply_modifications(env)
else:
env.update(extra_env)
if isinstance(extra_env, EnvironmentModifications):
extra_env.apply_modifications(current_environment)
elif extra_env is not None:
current_environment.update(extra_env)
if "_dump_env" in kwargs:
kwargs["_dump_env"].clear()
kwargs["_dump_env"].update(env)
if _dump_env is not None:
_dump_env.clear()
_dump_env.update(current_environment)
fail_on_error = kwargs.pop("fail_on_error", True)
ignore_errors = kwargs.pop("ignore_errors", ())
ignore_quotes = kwargs.pop("ignore_quotes", self.ignore_quotes)
timeout = kwargs.pop("timeout", None)
if ignore_quotes is None:
ignore_quotes = self.ignore_quotes
# If they just want to ignore one error code, make it a tuple.
if isinstance(ignore_errors, int):
ignore_errors = (ignore_errors,)
input = kwargs.pop("input", None)
output = kwargs.pop("output", None)
error = kwargs.pop("error", None)
if input is str:
raise ValueError("Cannot use `str` as input stream.")
@ -230,9 +261,15 @@ def streamify(arg, mode):
cmd_line_string = " ".join(escaped_cmd)
tty.debug(cmd_line_string)
result = None
try:
proc = subprocess.Popen(
cmd, stdin=istream, stderr=estream, stdout=ostream, env=env, close_fds=False
cmd,
stdin=istream,
stderr=estream,
stdout=ostream,
env=current_environment,
close_fds=False,
)
out, err = proc.communicate(timeout=timeout)
@ -248,9 +285,6 @@ def streamify(arg, mode):
long_msg += "\n" + result
raise ProcessError("Command exited with status %d:" % proc.returncode, long_msg)
return result
except OSError as e:
message = "Command: " + cmd_line_string
if " " in self.exe[0]:
@ -286,6 +320,8 @@ def streamify(arg, mode):
if close_istream:
istream.close()
return result
def __eq__(self, other):
return hasattr(other, "exe") and self.exe == other.exe