Skip to content
Back to top

# Source code for pyproject_api._via_fresh_subprocess

    from __future__ import annotations

    import os
    import sys
    from contextlib import contextmanager
    from subprocess import PIPE, Popen  # noqa: S404
    from threading import Thread
    from typing import IO, TYPE_CHECKING, Any, cast

    from ._frontend import CmdStatus, Frontend

    if TYPE_CHECKING:
        from collections.abc import Iterator
        from pathlib import Path

        from packaging.requirements import Requirement

    class SubprocessCmdStatus(CmdStatus, Thread):
        def __init__(self, process: Popen[str]) -> None:
            super().__init__()
            self.process = process
            self._out_err: tuple[str, str] | None = None
            self.start()

        def run(self) -> None:
            self._out_err = self.process.communicate()

        @property
        def done(self) -> bool:
            return self.process.returncode is not None

        def out_err(self) -> tuple[str, str]:
            return cast("tuple[str, str]", self._out_err)

    class SubprocessFrontend(Frontend):
        """A frontend that creates fresh subprocess at every call to communicate with the backend."""

        def __init__(
            self,
            root: Path,
            backend_paths: tuple[Path, ...],
            backend_module: str,
            backend_obj: str | None,
            requires: tuple[Requirement, ...],
        ) -> None:
            """
            Create a subprocess frontend.

            :param root: the root path to the built project
            :param backend_paths: paths that are available on the python path for the backend
            :param backend_module: module where the backend is located
            :param backend_obj: object within the backend module identifying the backend
            :param requires: seed requirements for the backend
            """
            super().__init__(root, backend_paths, backend_module, backend_obj, requires, reuse_backend=False)
            self.executable = sys.executable

        @contextmanager
        def _send_msg(self, cmd: str, result_file: Path, msg: str) -> Iterator[SubprocessCmdStatus]:  # noqa: ARG002
            env = os.environ.copy()
            backend = os.pathsep.join(str(i) for i in self._backend_paths).strip()
            if backend:
                env["PYTHONPATH"] = backend
            process = Popen(
                args=[self.executable, *self.backend_args],
                stdout=PIPE,
                stderr=PIPE,
                stdin=PIPE,
                universal_newlines=True,
                cwd=self._root,
                env=env,
            )
            cast("IO[str]", process.stdin).write(f"{os.linesep}{msg}{os.linesep}")
            yield SubprocessCmdStatus(process)

        def send_cmd(self, cmd: str, **kwargs: Any) -> tuple[Any, str, str]:
            """
            Send a command to the backend.

            :param cmd: the command to send
            :param kwargs: keyword arguments to the backend
            :return: a tuple of: backend response, standard output text, standard error text
            """
            return self._send(cmd, **kwargs)

    __all__ = ("SubprocessFrontend",)