def run_command()

in muss/utils/helpers.py [0:0]

• 46 lines of code
• 11 McCabe index (conditional complexity)

def run_command(cmd, mute=False):
    def readline_with_timeout(input_stream, timeout=0.1):
        """Avoids handing indefinitely when calling readline()
        https://stackoverflow.com/questions/10756383/timeout-on-subprocess-readline-in-python"""
        poll_obj = select.poll()
        poll_obj.register(input_stream, select.POLLIN)
        start = time.time()
        while (time.time() - start) < timeout:
            poll_result = poll_obj.poll(0)
            if poll_result:
                return input_stream.readline()
        return ''

    def get_available_output(input_stream):
        output = ''
        while True:
            line = readline_with_timeout(input_stream, timeout=0.1)
            if line == '':
                break
            output += line
        return output

    def read_and_print(input_stream, output_stream):
        output = get_available_output(input_stream)
        if not mute:
            print(output, file=output_stream, end='', flush=True)
        return output

    # Inspired from subprocess.run() source
    # HACK: shell=True is not secure
    with Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True, encoding='utf-8') as process:
        try:
            stdout = ''
            stderr = ''
            while True:
                stdout += read_and_print(process.stdout, sys.stdout)
                stderr += read_and_print(process.stderr, sys.stderr)
                if process.poll() is not None:
                    break
            # Read remaining output in case there is some
            stdout += read_and_print(process.stdout, sys.stdout)
            stderr += read_and_print(process.stderr, sys.stderr)
        except Exception:
            if mute:  # Print errors that could have been muted
                print(stderr, file=sys.stderr)
            print(get_available_output(process.stderr), file=sys.stderr)
            process.kill()
            process.wait()
            raise
        retcode = process.poll()
        if retcode:
            print(stderr, file=sys.stderr)
            raise CalledProcessError(retcode, process.args, output=stdout, stderr=stderr)
    return stdout.strip()