Python3 example to use Gio and GLib to run subprocesses and asynchronously read stdout/stderr

gio_subprocess.py

#!/usr/bin/env python3
from gi.repository import Gio, GLib
import shlex
import signal

priority = GLib.PRIORITY_DEFAULT


class ProcessLauncher:
    def run(self, cmd):
        self.cancellable = Gio.Cancellable()
        try:
            flags = Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_MERGE
            args = shlex.split(cmd)
            self.process = p = Gio.Subprocess.new(args, flags)
            p.wait_check_async(
                cancellable=self.cancellable,
                callback=self._on_finished
            )
            print('Started')
            stream = p.get_stdout_pipe()
            self.data_stream = Gio.DataInputStream.new(stream)
            self.queue_read()
        except GLib.GError as e:
            print(e)

    def queue_read(self):
        self.data_stream.read_line_async(
            io_priority=priority,
            cancellable=self.cancellable,
            callback=self._on_data
        )

    def cancel_read(self):
        print('Cancelling read')
        self.cancellable.cancel()

    def _on_finished(self, proc, results):
        print('Process finished')
        try:
            proc.wait_check_finish(results)
        except Exception as e:
            print(e)
        self.cancel_read()

    def _on_data(self, source, result):
        try:
            line, length = source.read_line_finish_utf8(result)
            if line:
                print(line)
        except GLib.GError as e:
            print(e)
            return
        self.queue_read()

    def stop(self):
        print('Stop')
        self.process.send_signal(signal.SIGTERM)

    def kill(self):
        print('Kill')
        self.cancel_read()
        self.process.send_signal(signal.SIGKILL)


if __name__ == '__main__':
    p = ProcessLauncher()
    m = GLib.MainLoop()
    GLib.idle_add(p.run, "/usr/bin/journalctl -f")
    GLib.timeout_add_seconds(priority=priority, interval=5, function=p.stop)
    m.run()

I am having trouble understanding whether queueing another read at the end of _on_data could be a problem if the subprocess sends a message while _on_data is processing a previous message. Would it be better to put the queue_read at the start of the handler?

Good question, i don't know GLib deep enough to respond !

My empirical conclusion is that it makes no difference where the queue_read goes. I have a subprocess emitting messages every second. I put a sleep command in _on_data that runs only the first time _on_data is called. Of course, no message gets processed while the handler is sleeping, but as soon as it wakes it quickly processes all the messages emitted during its slumber. Evidently, all the messages get buffered in stream until the handler pulls them out.

By the way, thank you for posting this code. It was very helpful in understanding these poorly documented features of Gio and GLib.

TBH i just found it somewhere and stored it exactly for the same reason. Thanks for the investigation !