Печать выходных строк стандартного вывода подпроцесса

Я создал простую функцию Python:

import subprocess
from io import TextIOWrapper


def run_shell_command(command: list, debug: bool = False):
    '''
    Run shell command

    :param command: Shell command
    :param debug: Debug mode
    :return: Result code and message
    '''
    try:
        process = subprocess.run(
            command, check=True, text=True, timeout=5,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT
        )
        if debug:
            for line in TextIOWrapper(process.stdout, encoding='utf-8'):
                print(line)
        message = 'Shell command executed sucessfully'
        return ({'code': 200, 'msg': message, 'stdout': process.stdout})
    except subprocess.CalledProcessError as e:
        return ({'code': 500, 'msg': e.output})


if __name__ == "__main__":
    command = run_shell_command(['ls', '-lah'], True)
    print(command)

Когда я запускаю его в режиме отладки, я получаю следующую ошибку:

Traceback (most recent call last):
  File "/tmp/command.py", line 28, in <module>
    command = run_shell_command(['ls', '-lah'], True)
  File "/tmp/command.py", line 19, in run_shell_command
    for line in TextIOWrapper(process.stdout, encoding = "utf-8"):
AttributeError: 'str' object has no attribute 'readable'

Запуская Python 3.9 на сервере Linux, мне было интересно, можете ли вы рассказать, в чем может быть проблема. При отключенной отладке я получаю правильный текстовый вывод. Спасибо за помощь.

Обновлено: Судя по комментариям ниже, исправить это довольно просто:

        if debug:
            print(process.stdout.rstrip())

Однако принятое решение IMO лучше по сравнению с исходным кодом.

🤔 А знаете ли вы, что...
С Python можно создавать кросс-платформенные приложения для Windows, macOS и Linux.


1
67
1

Ответ:

Решено

К сожалению, «простой» подход не позволяет справиться со всеми крайними случаями — вам нужно будет прочитать подпроцесс stdout во время его потоковой передачи, распечатать его и накопить в буфере, а также отслеживать время, чтобы вы могли тайм-аут правильно. Обратите внимание, что здесь тоже возможна ошибка (хотя и не очень опасная), если чтение 4096 байт заканчивается многострочным символом.

def run_shell_command(command: list, debug: bool = False, timeout: float = 5):
    """
    Run shell command

    :param command: Shell command
    :param debug: Debug mode
    :return: Result code and message
    """
    with subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    ) as process:
        start_time = time.time()
        output = b""
        while True:
            buf = process.stdout.read(4096)
            if debug:
                # could fail if `buf` happens to end in a multi-byte character
                print(buf.decode("utf-8", "ignore"))
            output += buf

            if time.time() - start_time > timeout:
                process.kill()
                message = "Shell command timed out"
                return {"code": 500, "msg": message, "stdout": output}

            if process.poll() is not None:
                break
    if process.returncode != 0:
        message = "Shell command failed"
        return {"code": 500, "msg": message, "stdout": output}

    message = "Shell command executed successfully"
    return {"code": 200, "msg": message, "stdout": output}

if __name__ == "__main__":
    command = run_shell_command(["ls", "-lah"], True)
    print(command)