Python 可执行文件 zip 中子进程的实时标准输出[重复]

Question

我编写了一个程序，它在新的子进程中运行 shell 命令，然后轮询它以获取 stdout 缓冲区中的内容，直到它完成运行或达到超时。这是一个更大项目的一部分，我用它来自动化 Docker 构建。

    def run(self, command: str) -> ShellProcess:
        log(f"Running command `{command}`...")

        start_time = time.time()

        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True, shell=True,
            bufsize=1,
            universal_newlines=True,
            env=self.environment
        )

        output = []

        while line := process.stdout.readline():
            output.append(line)
            print(line.strip())

            if self.timeout is not None:
                elapsed_time = time.time() - start_time
                if elapsed_time > self.timeout:
                    log(
                        f"{int(elapsed_time)} seconds have elapsed, exceeding the timeout... "
                        f"forcefully terminating process."
                    )
                    process.terminate()

        # Wait for the process to finish
        _, error = process.communicate()

        if error:
            log("Contents of stderr buffer:")
            print(error.strip())
            log("End stderr buffer")

        elapsed_time = int(time.time() - start_time)

        log(f"Process returned with status code {process.returncode} in {elapsed_time} seconds.")

        output = "".join(output)

        process = ShellProcess(
            command, output, error,process.returncode
        )

        return process

我正在通过将以下 Python 脚本作为子进程运行来进行测试：

import sys
import time
from datetime import datetime

duration = 3

if __name__ == "__main__":
    for i in range(duration):
        now = datetime.now()
        timestamp = now.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
        print(f"[{timestamp}]: {i+1}/{duration} seconds have elapsed...")
        time.sleep(1)

    print(f"This string is being printed to sys.stderr.", file=sys.stderr)
    print(f"So is this one...", file=sys.stderr)

如果我正常运行子进程，就会发生预期的结果：标准输出会实时逐行打印。

当我将此代码打包到 Python 可执行文件 zip 中时（使用 Bazel 的 Python 规则，但遵循与 https://docs.python.org/3/library/zipapp.html 概述的相同结构），不会打印任何内容进入控制台，直到进程完成或终止，此时写入 stdout 的每一行都会打印到控制台。测试代码中的时间戳是正确的：

[2024-01-16 11:29:47.169]: 1/3 seconds have elapsed...
[2024-01-16 11:29:48.169]: 2/3 seconds have elapsed...
[2024-01-16 11:29:49.174]: 3/3 seconds have elapsed...
This string is being printed to sys.stderr.
So is this one...

当代码打包在可执行 zip 中时，如何获得实时标准输出？

Answer 1

按照 AKX 的建议，设置

PYTHONUNBUFFERED=1

envvar 可以修复此问题。