"""Build frontend for PEP-517."""
from __future__ import annotations
import json
import sys
from abc import ABC, abstractmethod
from contextlib import contextmanager
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
from time import sleep
from typing import Any, Dict, Iterator, List, Literal, NamedTuple, NoReturn, Optional, TypedDict, cast
from zipfile import ZipFile
from packaging.requirements import Requirement
from pyproject_api._util import ensure_empty_dir
if sys.version_info >= (3, 11): # pragma: no cover (py311+)
import tomllib
else: # pragma: no cover (py311+)
import tomli as tomllib
_HERE = Path(__file__).parent
ConfigSettings = Optional[Dict[str, Any]]
[docs]class OptionalHooks(TypedDict, total=True):
"""A flag indicating if the backend supports the optional hook or not."""
get_requires_for_build_sdist: bool
prepare_metadata_for_build_wheel: bool
get_requires_for_build_wheel: bool
build_editable: bool
get_requires_for_build_editable: bool
prepare_metadata_for_build_editable: bool
class CmdStatus(ABC):
@property
@abstractmethod
def done(self) -> bool:
""":return: truthful when the command finished running"""
raise NotImplementedError
@abstractmethod
def out_err(self) -> tuple[str, str]:
""":return: standard output and standard error text"""
raise NotImplementedError
[docs]class RequiresBuildSdistResult(NamedTuple):
"""Information collected while acquiring the source distribution build dependencies."""
#: wheel build dependencies
requires: tuple[Requirement, ...]
#: backend standard output while acquiring the source distribution build dependencies
out: str
#: backend standard output while acquiring the source distribution build dependencies
err: str
[docs]class RequiresBuildWheelResult(NamedTuple):
"""Information collected while acquiring the wheel build dependencies."""
#: wheel build dependencies
requires: tuple[Requirement, ...]
#: backend standard output while acquiring the wheel build dependencies
out: str
#: backend standard error while acquiring the wheel build dependencies
err: str
[docs]class RequiresBuildEditableResult(NamedTuple):
"""Information collected while acquiring the wheel build dependencies."""
#: editable wheel build dependencies
requires: tuple[Requirement, ...]
#: backend standard output while acquiring the editable wheel build dependencies
out: str
#: backend standard error while acquiring the editable wheel build dependencies
err: str
[docs]class SdistResult(NamedTuple):
"""Information collected while building a source distribution."""
#: path to the built source distribution
sdist: Path
#: backend standard output while building the source distribution
out: str
#: backend standard output while building the source distribution
err: str
[docs]class WheelResult(NamedTuple):
"""Information collected while building a wheel."""
#: path to the built wheel artifact
wheel: Path
#: backend standard output while building the wheel
out: str
#: backend standard error while building the wheel
err: str
[docs]class EditableResult(NamedTuple):
"""Information collected while building an editable wheel."""
#: path to the built wheel artifact
wheel: Path
#: backend standard output while building the wheel
out: str
#: backend standard error while building the wheel
err: str
[docs]class BackendFailed(RuntimeError): # noqa: N818
"""An error of the build backend."""
def __init__(self, result: dict[str, Any], out: str, err: str) -> None:
super().__init__()
#: standard output collected while running the command
self.out = out
#: standard error collected while running the command
self.err = err
#: exit code of the command
self.code: int = result.get("code", -2)
#: the type of exception thrown
self.exc_type: str = result.get("exc_type", "missing Exception type")
#: the string representation of the exception thrown
self.exc_msg: str = result.get("exc_msg", "missing Exception message")
def __str__(self) -> str:
return (
f"packaging backend failed{'' if self.code is None else f' (code={self.code})'}, "
f"with {self.exc_type}: {self.exc_msg}\n{self.err}{self.out}"
).rstrip()
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}("
f"result=dict(code={self.code}, exc_type={self.exc_type!r},exc_msg={self.exc_msg!r}),"
f" out={self.out!r}, err={self.err!r})"
)
[docs]class Frontend(ABC):
"""Abstract base class for a pyproject frontend."""
#: backend key when the ``pyproject.toml`` does not specify it
LEGACY_BUILD_BACKEND: str = "setuptools.build_meta:__legacy__"
#: backend requirements when the ``pyproject.toml`` does not specify it
LEGACY_REQUIRES: tuple[Requirement, ...] = (Requirement("setuptools >= 40.8.0"), Requirement("wheel"))
def __init__( # noqa: PLR0913, PLR0917
self,
root: Path,
backend_paths: tuple[Path, ...],
backend_module: str,
backend_obj: str | None,
requires: tuple[Requirement, ...],
reuse_backend: bool = True, # noqa: FBT001, FBT002
) -> None:
"""
Create a new frontend.
:param root: the root path of the project
:param backend_paths: paths to provision as available to import from for the build backend
:param backend_module: the module where the backend lives
:param backend_obj: the backend object key (will be lookup up within the backend module)
:param requires: build requirements for the backend
:param reuse_backend: a flag indicating if the communication channel should be kept alive between messages
"""
self._root = root
self._backend_paths = backend_paths
self._backend_module = backend_module
self._backend_obj = backend_obj
self.requires: tuple[Requirement, ...] = requires
self._reuse_backend = reuse_backend
self._optional_hooks: OptionalHooks | None = None
[docs] @classmethod
def create_args_from_folder(
cls,
folder: Path,
) -> tuple[Path, tuple[Path, ...], str, str | None, tuple[Requirement, ...], bool]:
"""
Frontend creation arguments from a python project folder (thould have a ``pypyproject.toml`` file per PEP-518).
:param folder: the python project folder
:return: the frontend creation args
E.g., to create a frontend from a python project folder:
.. code:: python
frontend = Frontend(*Frontend.create_args_from_folder(project_folder))
"""
py_project_toml = folder / "pyproject.toml"
if py_project_toml.exists():
with py_project_toml.open("rb") as file_handler:
py_project = tomllib.load(file_handler)
build_system = py_project.get("build-system", {})
if "backend-path" in build_system:
backend_paths: tuple[Path, ...] = tuple(folder / p for p in build_system["backend-path"])
else:
backend_paths = ()
if "requires" in build_system:
requires: tuple[Requirement, ...] = tuple(Requirement(r) for r in build_system.get("requires"))
else:
requires = cls.LEGACY_REQUIRES
build_backend = build_system.get("build-backend", cls.LEGACY_BUILD_BACKEND)
else:
backend_paths = ()
requires = cls.LEGACY_REQUIRES
build_backend = cls.LEGACY_BUILD_BACKEND
paths = build_backend.split(":")
backend_module: str = paths[0]
backend_obj: str | None = paths[1] if len(paths) > 1 else None
return folder, backend_paths, backend_module, backend_obj, requires, True
@property
def backend(self) -> str:
""":return: backend key"""
return f"{self._backend_module}{f':{self._backend_obj}' if self._backend_obj else ''}"
@property
def backend_args(self) -> list[str]:
""":return: startup arguments for a backend"""
result: list[str] = [str(_HERE / "_backend.py"), str(self._reuse_backend), self._backend_module]
if self._backend_obj:
result.append(self._backend_obj)
return result
@property
def optional_hooks(self) -> OptionalHooks:
""":return: a dictionary indicating if the optional hook is supported or not"""
if self._optional_hooks is None:
result, _, __ = self._send("_optional_hooks")
self._optional_hooks = result
return self._optional_hooks
[docs] def get_requires_for_build_sdist(self, config_settings: ConfigSettings | None = None) -> RequiresBuildSdistResult:
"""
Get build requirements for a source distribution (per PEP-517).
:param config_settings: run arguments
:return: outcome
"""
if self.optional_hooks["get_requires_for_build_sdist"]:
result, out, err = self._send(cmd="get_requires_for_build_sdist", config_settings=config_settings)
else:
result, out, err = [], "", ""
if not isinstance(result, list) or not all(isinstance(i, str) for i in result):
self._unexpected_response("get_requires_for_build_sdist", result, "list of string", out, err)
return RequiresBuildSdistResult(tuple(Requirement(r) for r in cast(List[str], result)), out, err)
[docs] def get_requires_for_build_wheel(self, config_settings: ConfigSettings | None = None) -> RequiresBuildWheelResult:
"""
Get build requirements for a wheel (per PEP-517).
:param config_settings: run arguments
:return: outcome
"""
if self.optional_hooks["get_requires_for_build_wheel"]:
result, out, err = self._send(cmd="get_requires_for_build_wheel", config_settings=config_settings)
else:
result, out, err = [], "", ""
if not isinstance(result, list) or not all(isinstance(i, str) for i in result):
self._unexpected_response("get_requires_for_build_wheel", result, "list of string", out, err)
return RequiresBuildWheelResult(tuple(Requirement(r) for r in cast(List[str], result)), out, err)
[docs] def get_requires_for_build_editable(
self,
config_settings: ConfigSettings | None = None,
) -> RequiresBuildEditableResult:
"""
Get build requirements for an editable wheel build (per PEP-660).
:param config_settings: run arguments
:return: outcome
"""
if self.optional_hooks["get_requires_for_build_editable"]:
result, out, err = self._send(cmd="get_requires_for_build_editable", config_settings=config_settings)
else:
result, out, err = [], "", ""
if not isinstance(result, list) or not all(isinstance(i, str) for i in result):
self._unexpected_response("get_requires_for_build_editable", result, "list of string", out, err)
return RequiresBuildEditableResult(tuple(Requirement(r) for r in cast(List[str], result)), out, err)
def _check_metadata_dir(self, metadata_directory: Path) -> None:
if metadata_directory == self._root:
msg = f"the project root and the metadata directory can't be the same {self._root}"
raise RuntimeError(msg)
if metadata_directory.exists(): # start with fresh
ensure_empty_dir(metadata_directory)
metadata_directory.mkdir(parents=True, exist_ok=True)
[docs] def build_sdist(self, sdist_directory: Path, config_settings: ConfigSettings | None = None) -> SdistResult:
"""
Build a source distribution (per PEP-517).
:param sdist_directory: the folder where to build the source distribution
:param config_settings: build arguments
:return: source distribution build result
"""
sdist_directory.mkdir(parents=True, exist_ok=True)
basename, out, err = self._send(
cmd="build_sdist",
sdist_directory=sdist_directory,
config_settings=config_settings,
)
if not isinstance(basename, str):
self._unexpected_response("build_sdist", basename, str, out, err)
return SdistResult(sdist_directory / basename, out, err)
[docs] def build_wheel(
self,
wheel_directory: Path,
config_settings: ConfigSettings | None = None,
metadata_directory: Path | None = None,
) -> WheelResult:
"""
Build a wheel file (per PEP-517).
:param wheel_directory: the folder where to build the wheel
:param config_settings: build arguments
:param metadata_directory: wheel metadata folder
:return: wheel build result
"""
wheel_directory.mkdir(parents=True, exist_ok=True)
basename, out, err = self._send(
cmd="build_wheel",
wheel_directory=wheel_directory,
config_settings=config_settings,
metadata_directory=metadata_directory,
)
if not isinstance(basename, str):
self._unexpected_response("build_wheel", basename, str, out, err)
return WheelResult(wheel_directory / basename, out, err)
[docs] def build_editable(
self,
wheel_directory: Path,
config_settings: ConfigSettings | None = None,
metadata_directory: Path | None = None,
) -> EditableResult:
"""
Build an editable wheel file (per PEP-660).
:param wheel_directory: the folder where to build the editable wheel
:param config_settings: build arguments
:param metadata_directory: wheel metadata folder
:return: wheel build result
"""
wheel_directory.mkdir(parents=True, exist_ok=True)
basename, out, err = self._send(
cmd="build_editable",
wheel_directory=wheel_directory,
config_settings=config_settings,
metadata_directory=metadata_directory,
)
if not isinstance(basename, str):
self._unexpected_response("build_editable", basename, str, out, err)
return EditableResult(wheel_directory / basename, out, err)
def _unexpected_response( # noqa: PLR0913
self,
cmd: str,
got: Any,
expected_type: Any,
out: str,
err: str,
) -> NoReturn:
msg = f"{cmd!r} on {self.backend!r} returned {got!r} but expected type {expected_type!r}"
raise BackendFailed({"code": None, "exc_type": TypeError.__name__, "exc_msg": msg}, out, err)
@contextmanager
def _wheel_directory(self) -> Iterator[Path]: # noqa: PLR6301
with TemporaryDirectory() as wheel_directory:
yield Path(wheel_directory)
def _send(self, cmd: str, **kwargs: Any) -> tuple[Any, str, str]:
with NamedTemporaryFile(prefix=f"pep517_{cmd}-") as result_file_marker:
result_file = Path(result_file_marker.name).with_suffix(".json")
msg = json.dumps(
{
"cmd": cmd,
"kwargs": {k: (str(v) if isinstance(v, Path) else v) for k, v in kwargs.items()},
"result": str(result_file),
},
)
with self._send_msg(cmd, result_file, msg) as status:
while not status.done: # pragma: no branch
sleep(0.001) # wait a bit for things to happen
if result_file.exists():
try:
with result_file.open("rt") as result_handler:
result = json.load(result_handler)
finally:
result_file.unlink()
else:
result = {
"code": 1,
"exc_type": "RuntimeError",
"exc_msg": f"Backend response file {result_file} is missing",
}
out, err = status.out_err()
if "return" in result:
return result["return"], out, err
raise BackendFailed(result, out, err)
@abstractmethod
@contextmanager
def _send_msg(self, cmd: str, result_file: Path, msg: str) -> Iterator[CmdStatus]:
raise NotImplementedError