processes

Note that this function will consume stdout as soon as it’s available, so that you don’t need to worry about the buffer overflowing and such.

satella.processes.call_and_return_stdout(args, timeout=None, encoding=None, expected_return_code=None, **kwargs)

Call a process and return it’s stdout.

Everything in kwargs will be passed to subprocess.Popen

A bytes object will be returned if encoding is not defined, else stdout will be decoded according to specified encoding.

Deprecated since version Use: subprocess.check_output instead.

Parameters:
  • args (Union[str, List[str]]) – arguments to run the program with. Can be either a string or a list of strings.

  • timeout (Optional[Union[str, int]]) – amount of seconds to wait for the process result. If process does not complete within this time, it will be sent a SIGKILL. Can be also a time string. If left at default, ie. None, timeout won’t be considered at all.

  • encoding (Optional[str]) – encoding with which to decode stdout. If none is passed, it will be returned as a bytes object

  • expected_return_code (Optional[int]) – an expected return code of this process. 0 is the default. If process returns anything else, ProcessFailed will be raise. If left default (None) return code won’t be checked at all

Raises:
  • ProcessFailed – process’ result code was different from the requested

  • TimeoutError – timeout was specified and the process didn’t complete

Return type:

Union[bytes, str]

This is a subprocess helper:

satella.processes.read_nowait(process, output_list)

This spawns a thread to read given process’ stdout and append it to a list, in order to prevent buffer filling up completely.

To retrieve entire stdout after process finishes do

>>> ''.join(list)

This thread will terminate automatically after the process closes it’s stdout or finishes.

Parameters:
  • process (Popen) –

  • output_list (List[str]) –