python buffering in sys.stdout-CodePudding

Minimal example:

def main():
    for i in range(100):
        print("One line every 2s", end = "\n")
        time.sleep(2)
if __name__ == '__main__':
    with open("log.out", 'w') as out, open("log.err", 'w') as err:
        sys.stdout = out
        sys.stderr = err
        main()

I want the print statements to be written toe the stdout file after every line. I tried unsuccesfully:

python -u thisfile.py

In the bashrc (zshrc)


PYTHONUNBUFFERED = 1

open("log.out", 'wb', buffering = 0)

Changing the main function is no an option for the real case. A solution where I run the python file with bash and redirect errors and output is ok tough.

I know theres lots of questions like this, but none seem to work for me.

The solution should work ideally for python 3.8 and anything newer.

CodePudding user response：

You can use the subprocess module to run the script with the desired options and redirect the output and error streams to the log files:

import subprocess

with open("log.out", "w") as out, open("log.err", "w") as err:
    subprocess.run(["python", "-u", "thisfile.py"], stdout=out, stderr=err)

This will run the script in unbuffered mode (-u option) and redirect the output and error streams to the specified log files.

CodePudding user response：

If you use bash you can use:

[...]$ python -u main.py > >(tee -a log.out) 2> >(tee -a log.err >&2)

main.py

import sys
import time

def main():
    for i in range(100):
        print("One line every 2s", end = "\n")
        time.sleep(2)

if __name__ == '__main__':
    main()

Explanation: How do I write standard error to a file while using "tee" with a pipe?

CodePudding user response：

you could have a wrapper that automatically flushes stdout after every write. wrapper taken from this answer

import time
import sys

class Unbuffered(object):
    def __init__(self, stream):
        self.stream = stream

    def write(self, data):
        self.stream.write(data)
        self.stream.flush()

    def writelines(self, datas):
        self.stream.writelines(datas)
        self.stream.flush()

    def __getattr__(self, attr):
        return getattr(self.stream, attr)


def main():
    for i in range(100):
        print("One line every 2s", end = "\n")
        time.sleep(2)
if __name__ == '__main__':
    with open("log.out", 'w') as out, open("log.err", 'w') as err:
        sys.stdout = Unbuffered(out)
        sys.stderr = Unbuffered(err)
        main()