Add type-hints to which
and which_string
(#48554)
This commit is contained in:
parent
f8b2c65ddf
commit
6fac041d40
@ -10,7 +10,9 @@
|
|||||||
import sys
|
import sys
|
||||||
import sysconfig
|
import sysconfig
|
||||||
import warnings
|
import warnings
|
||||||
from typing import Dict, Optional, Sequence, Union
|
from typing import Optional, Sequence, Union
|
||||||
|
|
||||||
|
from typing_extensions import TypedDict
|
||||||
|
|
||||||
import archspec.cpu
|
import archspec.cpu
|
||||||
|
|
||||||
@ -18,13 +20,17 @@
|
|||||||
from llnl.util import tty
|
from llnl.util import tty
|
||||||
|
|
||||||
import spack.platforms
|
import spack.platforms
|
||||||
|
import spack.spec
|
||||||
import spack.store
|
import spack.store
|
||||||
import spack.util.environment
|
import spack.util.environment
|
||||||
import spack.util.executable
|
import spack.util.executable
|
||||||
|
|
||||||
from .config import spec_for_current_python
|
from .config import spec_for_current_python
|
||||||
|
|
||||||
QueryInfo = Dict[str, "spack.spec.Spec"]
|
|
||||||
|
class QueryInfo(TypedDict, total=False):
|
||||||
|
spec: spack.spec.Spec
|
||||||
|
command: spack.util.executable.Executable
|
||||||
|
|
||||||
|
|
||||||
def _python_import(module: str) -> bool:
|
def _python_import(module: str) -> bool:
|
||||||
@ -211,7 +217,9 @@ def _executables_in_store(
|
|||||||
):
|
):
|
||||||
spack.util.environment.path_put_first("PATH", [bin_dir])
|
spack.util.environment.path_put_first("PATH", [bin_dir])
|
||||||
if query_info is not None:
|
if query_info is not None:
|
||||||
query_info["command"] = spack.util.executable.which(*executables, path=bin_dir)
|
query_info["command"] = spack.util.executable.which(
|
||||||
|
*executables, path=bin_dir, required=True
|
||||||
|
)
|
||||||
query_info["spec"] = concrete_spec
|
query_info["spec"] = concrete_spec
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
@ -48,7 +48,13 @@
|
|||||||
import spack.version
|
import spack.version
|
||||||
from spack.installer import PackageInstaller
|
from spack.installer import PackageInstaller
|
||||||
|
|
||||||
from ._common import _executables_in_store, _python_import, _root_spec, _try_import_from_store
|
from ._common import (
|
||||||
|
QueryInfo,
|
||||||
|
_executables_in_store,
|
||||||
|
_python_import,
|
||||||
|
_root_spec,
|
||||||
|
_try_import_from_store,
|
||||||
|
)
|
||||||
from .clingo import ClingoBootstrapConcretizer
|
from .clingo import ClingoBootstrapConcretizer
|
||||||
from .config import spack_python_interpreter, spec_for_current_python
|
from .config import spack_python_interpreter, spec_for_current_python
|
||||||
|
|
||||||
@ -135,7 +141,7 @@ class BuildcacheBootstrapper(Bootstrapper):
|
|||||||
|
|
||||||
def __init__(self, conf) -> None:
|
def __init__(self, conf) -> None:
|
||||||
super().__init__(conf)
|
super().__init__(conf)
|
||||||
self.last_search: Optional[ConfigDictionary] = None
|
self.last_search: Optional[QueryInfo] = None
|
||||||
self.config_scope_name = f"bootstrap_buildcache-{uuid.uuid4()}"
|
self.config_scope_name = f"bootstrap_buildcache-{uuid.uuid4()}"
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@ -212,14 +218,14 @@ def _install_and_test(
|
|||||||
for _, pkg_hash, pkg_sha256 in item["binaries"]:
|
for _, pkg_hash, pkg_sha256 in item["binaries"]:
|
||||||
self._install_by_hash(pkg_hash, pkg_sha256, bincache_platform)
|
self._install_by_hash(pkg_hash, pkg_sha256, bincache_platform)
|
||||||
|
|
||||||
info: ConfigDictionary = {}
|
info: QueryInfo = {}
|
||||||
if test_fn(query_spec=abstract_spec, query_info=info):
|
if test_fn(query_spec=abstract_spec, query_info=info):
|
||||||
self.last_search = info
|
self.last_search = info
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
||||||
info: ConfigDictionary
|
info: QueryInfo
|
||||||
test_fn, info = functools.partial(_try_import_from_store, module), {}
|
test_fn, info = functools.partial(_try_import_from_store, module), {}
|
||||||
if test_fn(query_spec=abstract_spec_str, query_info=info):
|
if test_fn(query_spec=abstract_spec_str, query_info=info):
|
||||||
return True
|
return True
|
||||||
@ -232,7 +238,7 @@ def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
|||||||
return self._install_and_test(abstract_spec, bincache_platform, data, test_fn)
|
return self._install_and_test(abstract_spec, bincache_platform, data, test_fn)
|
||||||
|
|
||||||
def try_search_path(self, executables: Tuple[str], abstract_spec_str: str) -> bool:
|
def try_search_path(self, executables: Tuple[str], abstract_spec_str: str) -> bool:
|
||||||
info: ConfigDictionary
|
info: QueryInfo
|
||||||
test_fn, info = functools.partial(_executables_in_store, executables), {}
|
test_fn, info = functools.partial(_executables_in_store, executables), {}
|
||||||
if test_fn(query_spec=abstract_spec_str, query_info=info):
|
if test_fn(query_spec=abstract_spec_str, query_info=info):
|
||||||
self.last_search = info
|
self.last_search = info
|
||||||
@ -250,11 +256,11 @@ class SourceBootstrapper(Bootstrapper):
|
|||||||
|
|
||||||
def __init__(self, conf) -> None:
|
def __init__(self, conf) -> None:
|
||||||
super().__init__(conf)
|
super().__init__(conf)
|
||||||
self.last_search: Optional[ConfigDictionary] = None
|
self.last_search: Optional[QueryInfo] = None
|
||||||
self.config_scope_name = f"bootstrap_source-{uuid.uuid4()}"
|
self.config_scope_name = f"bootstrap_source-{uuid.uuid4()}"
|
||||||
|
|
||||||
def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
||||||
info: ConfigDictionary = {}
|
info: QueryInfo = {}
|
||||||
if _try_import_from_store(module, abstract_spec_str, query_info=info):
|
if _try_import_from_store(module, abstract_spec_str, query_info=info):
|
||||||
self.last_search = info
|
self.last_search = info
|
||||||
return True
|
return True
|
||||||
@ -289,7 +295,7 @@ def try_import(self, module: str, abstract_spec_str: str) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
def try_search_path(self, executables: Tuple[str], abstract_spec_str: str) -> bool:
|
def try_search_path(self, executables: Tuple[str], abstract_spec_str: str) -> bool:
|
||||||
info: ConfigDictionary = {}
|
info: QueryInfo = {}
|
||||||
if _executables_in_store(executables, abstract_spec_str, query_info=info):
|
if _executables_in_store(executables, abstract_spec_str, query_info=info):
|
||||||
self.last_search = info
|
self.last_search = info
|
||||||
return True
|
return True
|
||||||
@ -415,6 +421,7 @@ def ensure_executables_in_path_or_raise(
|
|||||||
current_bootstrapper.last_search["spec"],
|
current_bootstrapper.last_search["spec"],
|
||||||
current_bootstrapper.last_search["command"],
|
current_bootstrapper.last_search["command"],
|
||||||
)
|
)
|
||||||
|
assert cmd is not None, "expected an Executable"
|
||||||
cmd.add_default_envmod(
|
cmd.add_default_envmod(
|
||||||
spack.user_environment.environment_modifications_for_specs(
|
spack.user_environment.environment_modifications_for_specs(
|
||||||
concrete_spec, set_package_py_globals=False
|
concrete_spec, set_package_py_globals=False
|
||||||
|
@ -275,7 +275,7 @@ def _do_fake_install(pkg: "spack.package_base.PackageBase") -> None:
|
|||||||
fs.mkdirp(pkg.prefix.bin)
|
fs.mkdirp(pkg.prefix.bin)
|
||||||
fs.touch(os.path.join(pkg.prefix.bin, command))
|
fs.touch(os.path.join(pkg.prefix.bin, command))
|
||||||
if sys.platform != "win32":
|
if sys.platform != "win32":
|
||||||
chmod = which("chmod")
|
chmod = which("chmod", required=True)
|
||||||
chmod("+x", os.path.join(pkg.prefix.bin, command))
|
chmod("+x", os.path.join(pkg.prefix.bin, command))
|
||||||
|
|
||||||
# Install fake header file
|
# Install fake header file
|
||||||
|
@ -110,3 +110,11 @@ def test_which(tmpdir, monkeypatch):
|
|||||||
exe = ex.which("spack-test-exe")
|
exe = ex.which("spack-test-exe")
|
||||||
assert exe is not None
|
assert exe is not None
|
||||||
assert exe.path == path
|
assert exe.path == path
|
||||||
|
|
||||||
|
|
||||||
|
def test_construct_from_pathlib(mock_executable):
|
||||||
|
"""Tests that we can construct an executable from a pathlib.Path object"""
|
||||||
|
expected = "Hello world!"
|
||||||
|
path = mock_executable("hello", output=f"echo {expected}\n")
|
||||||
|
hello = ex.Executable(path)
|
||||||
|
assert expected in hello(output=str)
|
||||||
|
@ -7,7 +7,9 @@
|
|||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
from pathlib import Path, PurePath
|
from pathlib import Path, PurePath
|
||||||
from typing import Callable, Dict, Optional, Sequence, TextIO, Type, Union, overload
|
from typing import Callable, Dict, List, Optional, Sequence, TextIO, Type, Union, overload
|
||||||
|
|
||||||
|
from typing_extensions import Literal
|
||||||
|
|
||||||
import llnl.util.tty as tty
|
import llnl.util.tty as tty
|
||||||
|
|
||||||
@ -20,9 +22,9 @@
|
|||||||
class Executable:
|
class Executable:
|
||||||
"""Class representing a program that can be run on the command line."""
|
"""Class representing a program that can be run on the command line."""
|
||||||
|
|
||||||
def __init__(self, name: str) -> None:
|
def __init__(self, name: Union[str, Path]) -> None:
|
||||||
file_path = str(Path(name))
|
file_path = str(Path(name))
|
||||||
if sys.platform != "win32" and name.startswith("."):
|
if sys.platform != "win32" and isinstance(name, str) and name.startswith("."):
|
||||||
# pathlib strips the ./ from relative paths so it must be added back
|
# pathlib strips the ./ from relative paths so it must be added back
|
||||||
file_path = os.path.join(".", file_path)
|
file_path = os.path.join(".", file_path)
|
||||||
|
|
||||||
@ -338,10 +340,24 @@ def __str__(self):
|
|||||||
return " ".join(self.exe)
|
return " ".join(self.exe)
|
||||||
|
|
||||||
|
|
||||||
def which_string(*args, **kwargs):
|
@overload
|
||||||
"""Like ``which()``, but return a string instead of an ``Executable``."""
|
def which_string(
|
||||||
path = kwargs.get("path", os.environ.get("PATH", ""))
|
*args: str, path: Optional[Union[List[str], str]] = ..., required: Literal[True]
|
||||||
required = kwargs.get("required", False)
|
) -> str: ...
|
||||||
|
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def which_string(
|
||||||
|
*args: str, path: Optional[Union[List[str], str]] = ..., required: bool = ...
|
||||||
|
) -> Optional[str]: ...
|
||||||
|
|
||||||
|
|
||||||
|
def which_string(
|
||||||
|
*args: str, path: Optional[Union[List[str], str]] = None, required: bool = False
|
||||||
|
) -> Optional[str]:
|
||||||
|
"""Like ``which()``, but returns a string instead of an ``Executable``."""
|
||||||
|
if path is None:
|
||||||
|
path = os.environ.get("PATH", "")
|
||||||
|
|
||||||
if isinstance(path, list):
|
if isinstance(path, list):
|
||||||
paths = [Path(str(x)) for x in path]
|
paths = [Path(str(x)) for x in path]
|
||||||
@ -372,7 +388,6 @@ def add_extra_search_paths(paths):
|
|||||||
search_paths.insert(0, Path.cwd())
|
search_paths.insert(0, Path.cwd())
|
||||||
search_paths = add_extra_search_paths(search_paths)
|
search_paths = add_extra_search_paths(search_paths)
|
||||||
|
|
||||||
search_item = Path(search_item)
|
|
||||||
candidate_items = get_candidate_items(Path(search_item))
|
candidate_items = get_candidate_items(Path(search_item))
|
||||||
|
|
||||||
for candidate_item in candidate_items:
|
for candidate_item in candidate_items:
|
||||||
@ -385,29 +400,41 @@ def add_extra_search_paths(paths):
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
if required:
|
if required:
|
||||||
raise CommandNotFoundError("spack requires '%s'. Make sure it is in your path." % args[0])
|
raise CommandNotFoundError(f"spack requires '{args[0]}'. Make sure it is in your path.")
|
||||||
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def which(*args, **kwargs):
|
@overload
|
||||||
|
def which(
|
||||||
|
*args: str, path: Optional[Union[List[str], str]] = ..., required: Literal[True]
|
||||||
|
) -> Executable: ...
|
||||||
|
|
||||||
|
|
||||||
|
@overload
|
||||||
|
def which(
|
||||||
|
*args: str, path: Optional[Union[List[str], str]] = ..., required: bool = ...
|
||||||
|
) -> Optional[Executable]: ...
|
||||||
|
|
||||||
|
|
||||||
|
def which(
|
||||||
|
*args: str, path: Optional[Union[List[str], str]] = None, required: bool = False
|
||||||
|
) -> Optional[Executable]:
|
||||||
"""Finds an executable in the path like command-line which.
|
"""Finds an executable in the path like command-line which.
|
||||||
|
|
||||||
If given multiple executables, returns the first one that is found.
|
If given multiple executables, returns the first one that is found.
|
||||||
If no executables are found, returns None.
|
If no executables are found, returns None.
|
||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
*args (str): One or more executables to search for
|
*args: one or more executables to search for
|
||||||
|
path: the path to search. Defaults to ``PATH``
|
||||||
Keyword Arguments:
|
required: if set to True, raise an error if executable not found
|
||||||
path (list or str): The path to search. Defaults to ``PATH``
|
|
||||||
required (bool): If set to True, raise an error if executable not found
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Executable: The first executable that is found in the path
|
Executable: The first executable that is found in the path
|
||||||
"""
|
"""
|
||||||
exe = which_string(*args, **kwargs)
|
exe = which_string(*args, path=path, required=required)
|
||||||
return Executable(exe) if exe else None
|
return Executable(exe) if exe is not None else None
|
||||||
|
|
||||||
|
|
||||||
class ProcessError(spack.error.SpackError):
|
class ProcessError(spack.error.SpackError):
|
||||||
|
Loading…
Reference in New Issue
Block a user