Skip to content

shell

logger = get_logger() module-attribute

CommandFailed dataclass

Bases: Exception

Source code in shared/shell.py 
32
33
34
@dataclass
class CommandFailed(Exception):
    code: int

code: int class-attribute

consume_print(it)

It takes an iterator of strings and prints each line to the log

:param it: Iterator[str] :type it: Iterator[str]

Source code in shared/shell.py 
77
78
79
80
81
82
83
84
85
def consume_print(it: Iterator[str]):
    """
    It takes an iterator of strings and prints each line to the log

    :param it: Iterator[str]
    :type it: Iterator[str]
    """
    for line in it:
        logger.info(line)

execute_command(cmd, timeout=None, ignore_errors=False, **kwargs)

It runs a command and yields the output line by line

:param cmd: The command to execute :type cmd: Union[List[str], str] :param timeout: The maximum time to wait for the command to finish :type timeout: int :param ignore_errors: If True, don't raise an exception if the command fails, defaults to False :type ignore_errors: bool (optional)

Source code in shared/shell.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def execute_command(
    cmd: Union[List[str], str],
    timeout: int = None,
    ignore_errors: bool = False,
    **kwargs,
) -> Iterator[str]:
    """
    It runs a command and yields the output line by line

    :param cmd: The command to execute
    :type cmd: Union[List[str], str]
    :param timeout: The maximum time to wait for the command to finish
    :type timeout: int
    :param ignore_errors: If True, don't raise an exception if the command fails, defaults to False
    :type ignore_errors: bool (optional)
    """
    p = subprocess.Popen(
        cmd,
        shell=False,
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
        **kwargs
    )

    def run():
        for line in iter(p.stdout.readline, b''):
            line = line.rstrip().decode('utf-8')
            yield line

        p.stdout.close()
        code = p.wait()
        if code != 0 and not ignore_errors:
            raise CommandFailed(code)

    if timeout is not None:
        with with_timeout(p, timeout):
            yield from run()
    else:
        yield from run()

with_timeout(p, timeout)

It runs a function in a separate thread, and if the function doesn't return before the timeout, it kills the process

:param p: subprocess.Popen :type p: subprocess.Popen :param timeout: The maximum time to wait for the process to finish :type timeout: int

Source code in shared/shell.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@contextmanager
def with_timeout(p: subprocess.Popen, timeout: int):
    """
    It runs a function in a separate thread, and if the function doesn't return before the timeout, it kills the process

    :param p: subprocess.Popen
    :type p: subprocess.Popen
    :param timeout: The maximum time to wait for the process to finish
    :type timeout: int
    """
    def timerout(p: subprocess.Popen):
        p.kill()
        raise TimeoutError

    timer = Timer(timeout, timerout, [p])
    timer.start()
    yield
    timer.cancel()