Create a mock TextIOWrapper for using as a middleman for subprocess

Question

I am running an automation on a script and I want to evaluate the output of the shell instance before it gets written to a file.

bashCommand = "ls -a"
middleman = Middleman("Generic text", function_option, function)

process = subprocess.Popen(bashCommand.split(), stdout=middleman)
output, error = process.communicate()

The middleman class is intended to evaluate each line output of the bashCommand and then, if it finds a given text in the line, execute a function with a given setting.

Here is the Middleman class:

import io


class Middleman(io.TextIOBase):
    def __init__(self, target_text, target_settings, target_function):
        self.target_text = target_text
        self.target_function = target_function
        self.target_settings = target_settings

    def write(self, text, *args, **kwargs):
        if self.target_text in text:
            self.target_function(self.target_settings)
            print("Middleman:", "Function applied!")
            print("Middleman:", text)
        else:
            print("Middleman:", text)

    def fileno(self, *args, **kwargs):
        return 1

The problem is that subprocess stdout doesn't use .write() as the means to write to the TextIOWrapper, does anyone know how I can achieve the effect that I am looking for?

Thanks.

Answer 1

Instead of making Middleman a child of TextIOBase and passing it as stdout, you could use stdout=PIPE and have it call process.stdout.read().

class Middleman:
    def __init__(self, target_text, target_settings, target_function):
        self.target_text = target_text
        self.target_function = target_function
        self.target_settings = target_settings

    def read(self, text_stream):
        for text in text_stream:
            if self.target_text in text:
                self.target_function(self.target_settings)
                print("Middleman:", "Function applied!")
                print("Middleman:", text)
            else:
                print("Middleman:", text)


bashCommand = "ls -a"
middleman = Middleman("Generic text", function_option, function)

process = subprocess.Popen(bashCommand.split(), stdout=subprocess.PIPE)
middleman.read(process.stdout)
output, error = process.communicate()

You'd also have to handle accumulating the output. Since we consumed all the output in Middleman, output is probably empty.