Hide navigation sidebar
Hide table of contents sidebar
Back to top
Toggle table of contents sidebar

# Source code for pyproject_api._frontend

    """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]]

    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

    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

    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

    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

    class MetadataForBuildWheelResult(NamedTuple):
        """Information collected while acquiring the wheel metadata."""

        #: path to the wheel metadata
        metadata: Path
        #: backend standard output while generating the wheel metadata
        out: str
        #: backend standard output while generating the wheel metadata
        err: str

    class MetadataForBuildEditableResult(NamedTuple):
        """Information collected while acquiring the editable metadata."""

        #: path to the wheel metadata
        metadata: Path
        #: backend standard output while generating the editable wheel metadata
        out: str
        #: backend standard output while generating the editable wheel metadata
        err: str

    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

    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

    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

    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})"
            )

    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
            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

        @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

        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)

        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)

        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 prepare_metadata_for_build_wheel(
            self,
            metadata_directory: Path,
            config_settings: ConfigSettings | None = None,
        ) -> MetadataForBuildWheelResult | None:
            """
            Build wheel metadata (per PEP-517).

            :param metadata_directory: where to generate the metadata
            :param config_settings: build arguments
            :return: metadata generation result
            """
            self._check_metadata_dir(metadata_directory)
            basename: str | None = None
            if self.optional_hooks["prepare_metadata_for_build_wheel"]:
                basename, out, err = self._send(
                    cmd="prepare_metadata_for_build_wheel",
                    metadata_directory=metadata_directory,
                    config_settings=config_settings,
                )
            if basename is None:
                return None
            if not isinstance(basename, str):
                self._unexpected_response("prepare_metadata_for_build_wheel", basename, str, out, err)
            return MetadataForBuildWheelResult(metadata_directory / basename, 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)

        def prepare_metadata_for_build_editable(
            self,
            metadata_directory: Path,
            config_settings: ConfigSettings | None = None,
        ) -> MetadataForBuildEditableResult | None:
            """
            Build editable wheel metadata (per PEP-660).

            :param metadata_directory: where to generate the metadata
            :param config_settings: build arguments
            :return: metadata generation result
            """
            self._check_metadata_dir(metadata_directory)
            basename: str | None = None
            if self.optional_hooks["prepare_metadata_for_build_editable"]:
                basename, out, err = self._send(
                    cmd="prepare_metadata_for_build_editable",
                    metadata_directory=metadata_directory,
                    config_settings=config_settings,
                )
            if basename is None:
                return None
            if not isinstance(basename, str):
                self._unexpected_response("prepare_metadata_for_build_wheel", basename, str, out, err)
            result = metadata_directory / basename
            return MetadataForBuildEditableResult(result, out, err)

        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)

        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)

        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)

        def metadata_from_built(
            self,
            metadata_directory: Path,
            target: Literal["wheel", "editable"],
            config_settings: ConfigSettings | None = None,
        ) -> tuple[Path, str, str]:
            """
            Create metadata from building the wheel (use when the prepare endpoints are not present or don't work).

            :param metadata_directory: directory where to put the metadata
            :param target: the type of wheel metadata to build
            :param config_settings: config settings to pass in to the build endpoint
            :return:
            """
            hook = getattr(self, f"build_{target}")
            with self._wheel_directory() as wheel_directory:
                result: EditableResult | WheelResult = hook(wheel_directory, config_settings)
                wheel = result.wheel
                if not wheel.exists():
                    msg = f"missing wheel file return by backed {wheel!r}"
                    raise RuntimeError(msg)
                out, err = result.out, result.err
                extract_to = str(metadata_directory)
                basename = None
                with ZipFile(str(wheel), "r") as zip_file:
                    for name in zip_file.namelist():  # pragma: no branch
                        root = Path(name).parts[0]
                        if root.endswith(".dist-info"):
                            basename = root
                            zip_file.extract(name, extract_to)
            if basename is None:  # pragma: no branch
                msg = f"no .dist-info found inside generated wheel {wheel}"
                raise RuntimeError(msg)
            return metadata_directory / basename, out, err

        @contextmanager
        def _wheel_directory(self) -> Iterator[Path]:
            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