cardinal_pythonlib.subproc

Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).

This file is part of cardinal_pythonlib.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

class cardinal_pythonlib.subproc.AsynchronousFileReader(fd: BinaryIO, queue: Queue, encoding: str, line_terminators: List[str] | None = None, cmdargs: List[str] | None = None, suppress_decoding_errors: bool = True)[source]

Helper class to implement asynchronous reading of a file in a separate thread. Pushes read lines on a queue to be consumed in another thread.

Modified from https://stefaanlippens.net/python-asynchronous-subprocess-pipe-reading/.

Parameters:

  • fd – file-like object to read from

  • queue – queue to write to

  • encoding – encoding to use when reading from the file

  • line_terminators – valid line terminators

  • cmdargs – for display purposes only: command that produced/is producing the file-like object

  • suppress_decoding_errors – trap any UnicodeDecodeError?

eof() → bool[source]

Check whether there is no more content to expect.

run() → None[source]

Read lines and put them on the queue.

cardinal_pythonlib.subproc.check_call_process(args: List[str]) → None[source]

Logs the command arguments, then executes the command via subprocess.check_call().

cardinal_pythonlib.subproc.check_call_verbose(args: List[str], log_level: int | None = 20, **kwargs) → None[source]

Prints a copy/paste-compatible version of a command, then runs it.

Parameters:

  • args – command arguments

  • log_level – log level

Raises:

CalledProcessError –

cardinal_pythonlib.subproc.fail() → NoReturn[source]

Call when a child process has failed, and this will print an error message to stdout and execute sys.exit(1) (which will, in turn, call any atexit handler to kill children of this process).

cardinal_pythonlib.subproc.kill_child_processes() → None[source]

Kills children of this process that were registered in the processes variable.

Use with @atexit.register.

cardinal_pythonlib.subproc.mimic_user_input(args: List[str], source_challenge_response: List[Tuple[SubprocSource, str, str | SubprocCommand]], line_terminators: List[str] | None = None, print_stdout: bool = False, print_stderr: bool = False, print_stdin: bool = False, stdin_encoding: str | None = None, stdout_encoding: str | None = None, suppress_decoding_errors: bool = True, sleep_time_s: float = 0.1) → None[source]

Run an external command. Pretend to be a human by sending text to the subcommand (responses) when the external command sends us triggers (challenges).

This is a bit nasty.

Parameters:

  • args – command-line arguments

  • source_challenge_response – list of tuples of the format (challsrc, challenge, response); see below

  • line_terminators – valid line terminators

  • print_stdout –

  • print_stderr –

  • print_stdin –

  • stdin_encoding –

  • stdout_encoding –

  • suppress_decoding_errors – trap any UnicodeDecodeError?

  • sleep_time_s –

The (challsrc, challenge, response) tuples have this meaning:

  • challsrc: where is the challenge coming from? Must be one of the objects SOURCE_STDOUT or SOURCE_STDERR;

  • challenge: text of challenge

  • response: text of response (send to the subcommand’s stdin).

Example (modified from CorruptedZipReader):

from cardinal_pythonlib.subproc import *

SOURCE_FILENAME = "corrupt.zip"
TMP_DIR = "/tmp"
OUTPUT_FILENAME = "rescued.zip"

cmdargs = [
    "zip",  # Linux zip tool
    "-FF",  # or "--fixfix": "fix very broken things"
    SOURCE_FILENAME,  # input file
    "--temp-path", TMP_DIR,  # temporary storage path
    "--out", OUTPUT_FILENAME  # output file
]

# We would like to be able to say "y" automatically to
# "Is this a single-disk archive?  (y/n):"
# The source code (api.c, zip.c, zipfile.c), from
# ftp://ftp.info-zip.org/pub/infozip/src/ , suggests that "-q"
# should do this (internally "-q" sets "noisy = 0") - but in
# practice it doesn't work. This is a critical switch.
# Therefore we will do something very ugly, and send raw text via
# stdin.

ZIP_PROMPTS_RESPONSES = [
    (SOURCE_STDOUT, "Is this a single-disk archive?  (y/n): ", "y\n"),
    (SOURCE_STDOUT, " or ENTER  (try reading this split again): ", "q\n"),
    (SOURCE_STDERR,
     "zip: malloc.c:2394: sysmalloc: Assertion `(old_top == initial_top (av) "
     "&& old_size == 0) || ((unsigned long) (old_size) >= MINSIZE && "
     "prev_inuse (old_top) && ((unsigned long) old_end & (pagesize - 1)) "
     "== 0)' failed.", TERMINATE_SUBPROCESS),
]
ZIP_STDOUT_TERMINATORS = ["\n", "): "]

mimic_user_input(cmdargs,
                 source_challenge_response=ZIP_PROMPTS_RESPONSES,
                 line_terminators=ZIP_STDOUT_TERMINATORS,
                 print_stdout=show_zip_output,
                 print_stdin=show_zip_output)
cardinal_pythonlib.subproc.run_multiple_processes(args_list: List[List[str]], die_on_failure: bool = True, max_workers: int | None = None) → None[source]

Fire up multiple processes, and wait for them to finish.

Parameters:

  • args_list – command arguments for each process

  • die_on_failure – see wait_for_processes()

  • max_workers – Maximum simultaneous number of worker processes. Defaults to the total number of processes requested.