taucmdr.util module

Utility functions.

Handles system manipulation and status tasks, e.g. subprocess management or file creation.

taucmdr.util.add_error_stack(path)[source]
taucmdr.util.archive_toplevel(archive)[source]

Returns the name of the top-level directory in an archive.

Assumes that the archive file is rooted in a single top-level directory::
foo
/bar /baz

The top-level directory here is “foo” This routine will return stupid results for archives with multiple top-level elements.

Parameters:archive (str) – Path to archive file.
Raises:IOError – archive could not be read.
Returns:Directory name.
Return type:str
taucmdr.util.calculate_uid(parts)[source]

Create a new unique identifier.

Parameters:parts (list) – Ordered list of strings to include in the UID calcuation.
Returns:A string of hexadecimal digits uniquely calculated from parts.
Return type:str
taucmdr.util.camelcase(name)[source]

Converts a string to CamelCase.

Parameters:name (str) – String to convert.
Returns:name in CamelCase.
Return type:str
taucmdr.util.color_text(text, *args, **kwargs)[source]

Use termcolor.colored to colorize text.

Parameters:
  • text (str) – Text to colorize.
  • *args – Positional arguments to pass to termcolor.colored.
  • **kwargs – Keyword arguments to pass to termcolor.colored.
Returns:

The colorized text.
Return type:

str
taucmdr.util.copy_file(src, dest, show_progress=True)[source]

Works just like shutil.copy except with progress bars.

taucmdr.util.create_archive(fmt, dest, items, cwd=None, show_progress=True)[source]

Creates a new archive file in the specified format.

Parameters:
  • fmt (str) – Archive fmt, e.g. ‘zip’ or ‘tgz’.
  • dest (str) – Path to the archive file that will be created.
  • items (list) – Items (i.e. files or folders) to add to the archive.
  • cwd (str) – Current working directory while creating the archive.
taucmdr.util.create_subprocess(cmd, cwd=None, env=None, stdout=True, log=True, show_progress=False, error_buf=50, record_output=False)[source]

Create a subprocess.

See subprocess.Popen.

Parameters:
  • cmd (list) – Command and its command line arguments.
  • cwd (str) – If not None, change directory to cwd before creating the subprocess.
  • env (dict) – Environment variables to set or unset before launching cmd.
  • stdout (bool) – If True send subprocess stdout and stderr to this processes’ stdout.
  • log (bool) – If True send subprocess stdout and stderr to the debug log.
  • error_buf (int) – If non-zero, stdout is not already being sent, and return value is non-zero then send last error_buf lines of subprocess stdout and stderr to this processes’ stdout.
  • record_output (bool) – If True return output.
Returns:

Subprocess return code.
Return type:

int
taucmdr.util.download(src, dest, timeout=8)[source]

Downloads or copies files.

src may be a file path or URL. The destination folder will be created if it doesn’t exist. Download is via curl, wget, or Python’s urllib as appropriate.

Parameters:
  • src (str) – Path or URL to source file.
  • dest (str) – Path to file copy or download destination.
  • timeout (int) – Maximum time in seconds for the connection to the server. 0 for no timeout.
Raises:

IOError – File copy or download failed.
taucmdr.util.extract_archive(archive, dest, show_progress=True)[source]

Extracts archive file to dest.

Supports compressed and uncompressed tar archives. Destination folder will be created if it doesn’t exist.

Parameters:
  • archive (str) – Path to archive file to extract.
  • dest (str) – Destination folder.
Returns:

Full path to extracted files.
Return type:

str
Raises:

IOError – Failed to extract archive.
taucmdr.util.get_binary_linkage(cmd)[source]
taucmdr.util.get_command_output(cmd)[source]

Return the possibly cached output of a command.

Just subprocess.check_output with a cache. Subprocess stderr is always sent to subprocess stdout.

Parameters:cmd (list) – Command and its command line arguments.
Raises:subprocess.CalledProcessError – return code was non-zero.
Returns:Subprocess output.
Return type:str
taucmdr.util.hline(title, *args, **kwargs)[source]

Build a colorful horizontal rule for console output.

Uses logger.LINE_WIDTH to generate a string of ‘=’ characters as wide as the terminal. title is included in the string near the left of the horizontal line.

Parameters:
  • title (str) – Text to put on the horizontal rule.
  • *args – Positional arguments to pass to termcolor.colored.
  • **kwargs – Keyword arguments to pass to termcolor.colored.
Returns:

The horizontal rule.
Return type:

str
taucmdr.util.human_size(num, suffix='B')[source]

Converts a byte count to human readable units.

Parameters:
  • num (int) – Number to convert.
  • suffix (str) – Unit suffix, e.g. ‘B’ for bytes.
Returns:

num as a human readable string.
Return type:

str
taucmdr.util.is_clean_container(obj)[source]

Recursively checks a container for bytes, bytearray and memory view objects in keys and values. Care must be taken to avoid infinite loops caused by cycles in a dictionary with cycles.

Parameters:obj (any) – Container to be checked for disallowed binary data
Returns:Whether or not bad binary entries were found
Return type:bool
taucmdr.util.is_url(url)[source]

Check if url is a URL.

Parameters:url (str) – String to check
Returns:True if url is a URL, False otherwise.
Return type:bool
taucmdr.util.mkdirp(*args)[source]

Creates a directory and all its parents.

Works just like mkdir -p.

Parameters:*args – Paths to create.
taucmdr.util.mkdtemp(*args, **kwargs)[source]

Like tempfile.mkdtemp but directory will be recursively deleted when program exits.

taucmdr.util.page_output(output_string: str)[source]

Pipe string to a pager.

If PAGER is an environment then use that as pager, otherwise use less.

Parameters:output_string (str) – String to put output.
taucmdr.util.parse_bool(value, additional_true=None, additional_false=None)[source]

Parses a value to a boolean value.

If value is a string try to interpret it as a bool: * [‘1’, ‘t’, ‘y’, ‘true’, ‘yes’, ‘on’] ==> True * [‘0’, ‘f’, ‘n’, ‘false’, ‘no’, ‘off’] ==> False Otherwise raise TypeError.

Parameters:
  • value – value to parse to a boolean.
  • additional_true (list) – optional additional string values that stand for True.
  • additional_false (list) – optional additional string values that stand for False.
Returns:

True if value is true, False if value is false.
Return type:

bool
Raises:

ValueError – value does not parse.
taucmdr.util.path_accessible(path, mode='r')[source]

Check if a file or directory exists and is accessible.

Files are checked by attempting to open them with the given mode. Directories are checked by testing their access bits only, which may fail for some filesystems which may have permissions semantics beyond the usual POSIX permission-bit model. We’ll fix this if it becomes a problem.

Parameters:
  • path (str) – Path to file or directory to check.
  • mode (str) – File access mode to test, e.g. ‘r’ or ‘rw’
Returns:

True if the file exists and can be opened in the specified mode, False otherwise.
taucmdr.util.rmtree(path, ignore_errors=False, onerror=None, attempts=5)[source]

Wrapper around shutil.rmtree to work around stale or slow NFS directories.

Tries repeatedly to recursively remove path and sleeps between attempts.

Parameters:
  • path (str) – A directory but not a symbolic link to a directory.
  • ignore_errors (bool) – If True then errors resulting from failed removals will be ignored. If False or omitted, such errors are handled by calling a handler specified by onerror or, if that is omitted, they raise an exception.
  • onerror – Callable that accepts three parameters: function, path, and excinfo. See :any:shutil.rmtree.
  • attempts (int) – Number of times to repeat shutil.rmtree before giving up.
taucmdr.util.tmpfs_prefix()[source]

Path to a uniquely named directory in a temporary filesystem, ideally a ramdisk.

/dev/shm is the preferred tmpfs, but if it’s unavailable or mounted with noexec then fall back to tempfile.gettempdir(), which is usually /tmp. If that filesystem is also unavailable then use the filesystem prefix of the highest writable storage container.

An attempt is made to ensure that there is at least 2GiB of free space in the selected filesystem.

Returns:Path to a uniquely-named directory in the temporary filesystem. The directory and all its contents will be deleted when the program exits if it installs correctly.
Return type:str
taucmdr.util.umask(new_mask)[source]

Context manager to temporarily set the process umask.

Parameters:new_mask – The argument to os.umask.
taucmdr.util.uncolor_text(text)[source]

Remove color control chars from a string.

Parameters:text (str) – Text to colorize.
Returns:The text without control chars.
Return type:str
taucmdr.util.which(program, use_cached=True)[source]

Returns the full path to a program command.

Program must exist and be executable. Searches the system PATH and the current directory. Caches the result.

Parameters:
  • program (str) – program to find.
  • use_cached (bool) – If False then don’t use cached results.
Returns:

Full path to program or None if program can’t be found.
Return type:

str