[Python] Why is stdout PIPE readline() not waiting for a newline character?

Discussão em 'Python' iniciado por Stack, Abril 12, 2025.

  1. Stack

    Stack Membro Participativo

    Why is print(line) in read_stdout printing every character instead of the entire line? I am expecting it to add to the queue once the newline character is reached, but it's just putting every character in the queue instead.

    plugin_test.py:

    from subprocess import PIPE, Popen
    from threading import Thread
    from queue import Queue, Empty
    import re
    import os
    import sys

    def read_stdout(stdout, data_q):
    for line in stdout.readline():
    print(line)
    data_q.put(line)
    stdout.close()

    class Plugin():
    name = ''

    def __init__(self, **kwargs) -> None:
    self.process = None

    def run(self) -> bool:
    try:
    self.process = Popen(
    [sys.executable, 'test.py'],
    stdin=PIPE,
    stdout=PIPE,
    text=True)
    except Exception as exc:
    print("Could not create plugin process {}. {}".format(self.name, exc))
    return False

    self.data_q = Queue()
    self.read_t = Thread(target=read_stdout, args=(self.process.stdout, self.data_q))
    self.read_t.daemon = True
    self.read_t.start()

    try:
    startup = self.data_q.get(timeout=10)
    except Empty:
    print("Plugin took to long to load {}.".format(self.name))
    self.stop()
    return False
    else:
    if 'Error' in startup:
    print("Could not load plugin {}. {}".format(self.name, startup))
    self.stop()
    return False
    elif '100%' in startup:
    print("Plugin \"{}\" started.".format(self.name))

    return True

    def write(self, data : str) -> None:
    self.process.stdin.write(data)
    self.process.stdin.flush()

    def read(self) -> str:
    try:
    data = self.data_q.get()
    print("Got data ::{}::".format(data))
    except Exception as exc:
    print("Error reading data from plugin '{}'".format(exc))

    def stop(self):
    self.process.terminate()
    self.process.wait()


    if __name__ == '__main__':
    plugin = Plugin()
    plugin.run()


    test.py:

    print("this is a test")
    print("this is a test2")
    print("100%")


    root@osboxes# python plugin_test.py

    t
    h
    i
    s

    i
    s

    a

    t
    e
    s
    t

    Continue reading...

Compartilhe esta Página