Edit File: _impl.py
import json import os import sys import tempfile from contextlib import contextmanager from os.path import abspath from os.path import join as pjoin from subprocess import STDOUT, check_call, check_output from typing import TYPE_CHECKING, Any, Iterator, Mapping, Optional, Sequence from ._in_process import _in_proc_script_path if TYPE_CHECKING: from typing import Protocol class SubprocessRunner(Protocol): """A protocol for the subprocess runner.""" def __call__( self, cmd: Sequence[str], cwd: Optional[str] = None, extra_environ: Optional[Mapping[str, str]] = None, ) -> None: ... def write_json(obj: Mapping[str, Any], path: str, **kwargs) -> None: with open(path, "w", encoding="utf-8") as f: json.dump(obj, f, **kwargs) def read_json(path: str) -> Mapping[str, Any]: with open(path, encoding="utf-8") as f: return json.load(f) class BackendUnavailable(Exception): """Will be raised if the backend cannot be imported in the hook process.""" def __init__( self, traceback: str, message: Optional[str] = None, backend_name: Optional[str] = None, backend_path: Optional[Sequence[str]] = None, ) -> None: # Preserving arg order for the sake of API backward compatibility. self.backend_name = backend_name self.backend_path = backend_path self.traceback = traceback super().__init__(message or "Error while importing backend") class HookMissing(Exception): """Will be raised on missing hooks (if a fallback can't be used).""" def __init__(self, hook_name: str) -> None: super().__init__(hook_name) self.hook_name = hook_name class UnsupportedOperation(Exception): """May be raised by build_sdist if the backend indicates that it can't.""" def __init__(self, traceback: str) -> None: self.traceback = traceback def default_subprocess_runner( cmd: Sequence[str], cwd: Optional[str] = None, extra_environ: Optional[Mapping[str, str]] = None, ) -> None: """The default method of calling the wrapper subprocess. This uses :func:`subprocess.check_call` under the hood. """ env = os.environ.copy() if extra_environ: env.update(extra_environ) check_call(cmd, cwd=cwd, env=env) def quiet_subprocess_runner( cmd: Sequence[str], cwd: Optional[str] = None, extra_environ: Optional[Mapping[str, str]] = None, ) -> None: """Call the subprocess while suppressing output. This uses :func:`subprocess.check_output` under the hood. """ env = os.environ.copy() if extra_environ: env.update(extra_environ) check_output(cmd, cwd=cwd, env=env, stderr=STDOUT) def norm_and_check(source_tree: str, requested: str) -> str: """Normalise and check a backend path. Ensure that the requested backend path is specified as a relative path, and resolves to a location under the given source tree. Return an absolute version of the requested path. """ if os.path.isabs(requested): raise ValueError("paths must be relative") abs_source = os.path.abspath(source_tree) abs_requested = os.path.normpath(os.path.join(abs_source, requested)) # We have to use commonprefix for Python 2.7 compatibility. So we # normalise case to avoid problems because commonprefix is a character # based comparison :-( norm_source = os.path.normcase(abs_source) norm_requested = os.path.normcase(abs_requested) if os.path.commonprefix([norm_source, norm_requested]) != norm_source: raise ValueError("paths must be inside source tree") return abs_requested class BuildBackendHookCaller: """A wrapper to call the build backend hooks for a source directory.""" def __init__( self, source_dir: str, build_backend: str, backend_path: Optional[Sequence[str]] = None, runner: Optional["SubprocessRunner"] = None, python_executable: Optional[str] = None, ) -> None: """ :param source_dir: The source directory to invoke the build backend for :param build_backend: The build backend spec :param backend_path: Additional path entries for the build backend spec :param runner: The :ref:`subprocess runner <Subprocess Runners>` to use :param python_executable: The Python executable used to invoke the build backend """ if runner is None: runner = default_subprocess_runner self.source_dir = abspath(source_dir) self.build_backend = build_backend if backend_path: backend_path = [norm_and_check(self.source_dir, p) for p in backend_path] self.backend_path = backend_path self._subprocess_runner = runner if not python_executable: python_executable = sys.executable self.python_executable = python_executable @contextmanager def subprocess_runner(self, runner: "SubprocessRunner") -> Iterator[None]: """A context manager for temporarily overriding the default :ref:`subprocess runner <Subprocess Runners>`. :param runner: The new subprocess runner to use within the context. .. code-block:: python hook_caller = BuildBackendHookCaller(...) with hook_caller.subprocess_runner(quiet_subprocess_runner): ... """ prev = self._subprocess_runner self._subprocess_runner = runner try: yield finally: self._subprocess_runner = prev def _supported_features(self) -> Sequence[str]: """Return the list of optional features supported by the backend.""" return self._call_hook("_supported_features", {}) def get_requires_for_build_wheel( self, config_settings: Optional[Mapping[str, Any]] = None, ) -> Sequence[str]: """Get additional dependencies required for building a wheel. :param config_settings: The configuration settings for the build backend :returns: A list of :pep:`dependency specifiers <508>`. .. admonition:: Fallback If the build backend does not defined a hook with this name, an empty list will be returned. """ return self._call_hook( "get_requires_for_build_wheel", {"config_settings": config_settings} ) def prepare_metadata_for_build_wheel( self, metadata_directory: str, config_settings: Optional[Mapping[str, Any]] = None, _allow_fallback: bool = True, ) -> str: """Prepare a ``*.dist-info`` folder with metadata for this project. :param metadata_directory: The directory to write the metadata to :param config_settings: The configuration settings for the build backend :param _allow_fallback: Whether to allow the fallback to building a wheel and extracting the metadata from it. Should be passed as a keyword argument only. :returns: Name of the newly created subfolder within ``metadata_directory``, containing the metadata. .. admonition:: Fallback If the build backend does not define a hook with this name and ``_allow_fallback`` is truthy, the backend will be asked to build a wheel via the ``build_wheel`` hook and the dist-info extracted from that will be returned. """ return self._call_hook( "prepare_metadata_for_build_wheel", { "metadata_directory": abspath(metadata_directory), "config_settings": config_settings, "_allow_fallback": _allow_fallback, }, ) def build_wheel( self, wheel_directory: str, config_settings: Optional[Mapping[str, Any]] = None, metadata_directory: Optional[str] = None, ) -> str: """Build a wheel from this project. :param wheel_directory: The directory to write the wheel to :param config_settings: The configuration settings for the build backend :param metadata_directory: The directory to reuse existing metadata from :returns: The name of the newly created wheel within ``wheel_directory``. .. admonition:: Interaction with fallback If the ``build_wheel`` hook was called in the fallback for :meth:`prepare_metadata_for_build_wheel`, the build backend would not be invoked. Instead, the previously built wheel will be copied to ``wheel_directory`` and the name of that file will be returned. """ if metadata_directory is not None: metadata_directory = abspath(metadata_directory) return self._call_hook( "build_wheel", { "wheel_directory": abspath(wheel_directory), "config_settings": config_settings, "metadata_directory": metadata_directory, }, ) def get_requires_for_build_editable( self, config_settings: Optional[Mapping[str, Any]] = None, ) -> Sequence[str]: """Get additional dependencies required for building an editable wheel. :param config_settings: The configuration settings for the build backend :returns: A list of :pep:`dependency specifiers <508>`. .. admonition:: Fallback If the build backend does not defined a hook with this name, an empty list will be returned. """ return self._call_hook( "get_requires_for_build_editable", {"config_settings": config_settings} ) def prepare_metadata_for_build_editable( self, metadata_directory: str, config_settings: Optional[Mapping[str, Any]] = None, _allow_fallback: bool = True, ) -> Optional[str]: """Prepare a ``*.dist-info`` folder with metadata for this project. :param metadata_directory: The directory to write the metadata to :param config_settings: The configuration settings for the build backend :param _allow_fallback: Whether to allow the fallback to building a wheel and extracting the metadata from it. Should be passed as a keyword argument only. :returns: Name of the newly created subfolder within ``metadata_directory``, containing the metadata. .. admonition:: Fallback If the build backend does not define a hook with this name and ``_allow_fallback`` is truthy, the backend will be asked to build a wheel via the ``build_editable`` hook and the dist-info extracted from that will be returned. """ return self._call_hook( "prepare_metadata_for_build_editable", { "metadata_directory": abspath(metadata_directory), "config_settings": config_settings, "_allow_fallback": _allow_fallback, }, ) def build_editable( self, wheel_directory: str, config_settings: Optional[Mapping[str, Any]] = None, metadata_directory: Optional[str] = None, ) -> str: """Build an editable wheel from this project. :param wheel_directory: The directory to write the wheel to :param config_settings: The configuration settings for the build backend :param metadata_directory: The directory to reuse existing metadata from :returns: The name of the newly created wheel within ``wheel_directory``. .. admonition:: Interaction with fallback If the ``build_editable`` hook was called in the fallback for :meth:`prepare_metadata_for_build_editable`, the build backend would not be invoked. Instead, the previously built wheel will be copied to ``wheel_directory`` and the name of that file will be returned. """ if metadata_directory is not None: metadata_directory = abspath(metadata_directory) return self._call_hook( "build_editable", { "wheel_directory": abspath(wheel_directory), "config_settings": config_settings, "metadata_directory": metadata_directory, }, ) def get_requires_for_build_sdist( self, config_settings: Optional[Mapping[str, Any]] = None, ) -> Sequence[str]: """Get additional dependencies required for building an sdist. :returns: A list of :pep:`dependency specifiers <508>`. """ return self._call_hook( "get_requires_for_build_sdist", {"config_settings": config_settings} ) def build_sdist( self, sdist_directory: str, config_settings: Optional[Mapping[str, Any]] = None, ) -> str: """Build an sdist from this project. :returns: The name of the newly created sdist within ``wheel_directory``. """ return self._call_hook( "build_sdist", { "sdist_directory": abspath(sdist_directory), "config_settings": config_settings, }, ) def _call_hook(self, hook_name: str, kwargs: Mapping[str, Any]) -> Any: extra_environ = {"_PYPROJECT_HOOKS_BUILD_BACKEND": self.build_backend} if self.backend_path: backend_path = os.pathsep.join(self.backend_path) extra_environ["_PYPROJECT_HOOKS_BACKEND_PATH"] = backend_path with tempfile.TemporaryDirectory() as td: hook_input = {"kwargs": kwargs} write_json(hook_input, pjoin(td, "input.json"), indent=2) # Run the hook in a subprocess with _in_proc_script_path() as script: python = self.python_executable self._subprocess_runner( [python, abspath(str(script)), hook_name, td], cwd=self.source_dir, extra_environ=extra_environ, ) data = read_json(pjoin(td, "output.json")) if data.get("unsupported"): raise UnsupportedOperation(data.get("traceback", "")) if data.get("no_backend"): raise BackendUnavailable( data.get("traceback", ""), message=data.get("backend_error", ""), backend_name=self.build_backend, backend_path=self.backend_path, ) if data.get("hook_missing"): raise HookMissing(data.get("missing_hook_name") or hook_name) return data["return_val"]
Back to File Manager