Function _stream_python_script

helper_api.py:493–557

(
    script: str,
    success_message: str,
    failure_message: str,
    *,
    env_overrides: dict[str, str] | None = None,
    cleanup_paths: list[Path] | None = None,
)

def _stream_python_script(
    script: str,
    success_message: str,
    failure_message: str,
    *,
    env_overrides: dict[str, str] | None = None,
    cleanup_paths: list[Path] | None = None,
):
    cleanup_paths = cleanup_paths or []
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".py", dir=PROJECT_DIR, encoding="utf-8", mode="w") as temp_script:
            temp_script.write(script)
            temp_path = Path(temp_script.name)

        command = [sys.executable, str(temp_path)]
        yield f"$ {_format_command(command)}\n"

        env = os.environ.copy()
        env["PYTHONUNBUFFERED"] = "1"
        env["PYTHONIOENCODING"] = "utf-8"
        if env_overrides:
            env.update(env_overrides)

        try:
            process = subprocess.Popen(
                command,
                cwd=PROJECT_DIR,
                env=env,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                encoding="utf-8",
                errors="replace",
                bufsize=1,
            )
        except OSError as exc:
            yield f"Failed to start process: {exc}\n"
            yield _stream_status(False, failure_message)
            return

        if process.stdout is not None:
            for line in process.stdout:
                yield line

        returncode = process.wait()
        if returncode != 0:
            yield f"\nProcess exited with code {returncode}.\n"
            yield _stream_status(False, failure_message)
            return

        yield _stream_status(True, success_message)
    finally:
        if temp_path and temp_path.exists():
            try:
                temp_path.unlink()
            except OSError:
                pass