zensols.util package

Submodules

zensols.util.clip module

Provides access to the system clipboard. Currently only macOS is supported.

class zensols.util.clip.Clipboard

Bases: object

A utility class that provides access to the system clipboard.

read()

Read text from the system clipboard and return it.

Return type:

str

write(text)

Copy text to the system clipboard.

zensols.util.executor module

Simplifies external process handling.

class zensols.util.executor.ExecutableFinder(path_var=None, raise_on_missing=True)

Bases: object

Searches for an executable binary in the search path. The default search path (path_var) is set to the operating system’s PATH environment variable.

__init__(path_var=None, raise_on_missing=True)

find(name)

Like find_all(), but returns only the first found executable.

Raises:

OSError – if executable name is not found

Return type:

Path

find_all(name)

Return matches of executable binary name, if any.

Return type:

Iterable[Path]

path_var: str = None

The string that gives a delimited list of directories to search for an executable. This defaults to the PATH variable separated by the path separator (i.e. : in UNIX/Linux).

raise_on_missing: bool = True

Whether to raise errors when executables are not found.

property search_path: Tuple[Path, ...]

The search path dervied from path_var.

class zensols.util.executor.Executor(logger, dry_run=False, check_exit_value=0, timeout=None, async_proc=False, working_dir=None)

Bases: object

Run a process and log output. The process is run in the foreground by default, or background. If the later, a process object is returned from run().

__init__(logger, dry_run=False, check_exit_value=0, timeout=None, async_proc=False, working_dir=None)

async_proc: bool = False

If True, return a process from run(), which calls wait().

check_exit_value: int = 0

Compare and raise an exception if the exit value of the process is not this number, or None to not check.

dry_run: bool = False

If True do not do anything, just log as if it were to act/do something.

logger: Logger

The client logger used to log output of the process.

run(cmd)

Run a commmand.

Parameters:

cmd (Union[str, Iterable[str], Path]) – either one string, a sequence of arguments or a path (see subprocess.Popen)

Return type:

Union[Popen, int, None]

Returns:

the process if async_proc is True, otherwise, the exit status of the subprocess

timeout: int = None

The wait timeout in wait().

wait(proc)

Wait for process proc to end and return the processes exit value.

Return type:

int

working_dir: Path = None

Used as the cwd when creating Popen.

zensols.util.fail module

General utility classes to measuring time it takes to do work, to logging and fork/exec operations.

exception zensols.util.fail.APIError

Bases: Exception

Base exception from which almost all library raised errors extend.

__annotations__ = {}

__module__ = 'zensols.util.fail'

__weakref__

list of weak references to the object

class zensols.util.fail.Failure(exception=None, thrower=None, traceback=None, message=None)

Bases: Writable

Contains information for failures as caught exceptions used by programatic methods.

__init__(exception=None, thrower=None, traceback=None, message=None)

asdict()

Return the content of the object as a dictionary.

Return type:

Dict[str, Any]

exception: Exception = None

The exception that was generated.

message: str = None

A high level explaination of what went wrong

print_stack(writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Print the stack trace of the exception that caused the failure.

rethrow()

Raises exception.

thrower: Any = None

The instance of the class that is throwing the exception.

traceback: Field = None

The stack trace.

property traceback_str: str

The stack trace as a string.

:see print_stack()

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write the contents of this instance to writer using indention depth.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

zensols.util.hasher module

Utilities for hashing text.

class zensols.util.hasher.Hasher(short=True, decode='hex')

Bases: object

A utility class that creates hexadecimal hash values for text.

See:

hashlib

__init__(short=True, decode='hex')

decode: str = 'hex'

The decoded output representation. Choices are those b2a_* functions in bin2ascii such as hex, base64, etc.

reset()

short: bool = True

Whether or not to generate a 32-bit or 64-bit key.

update(text)

Update the hash from the text provided.

zensols.util.log module

Utility classes and context managers around logging.

class zensols.util.log.LogConfigurer(logger=<RootLogger root (WARNING)>, log_format='%(asctime)s %(levelname)s %(message)s', level=None)

Bases: object

Configure logging to go to a file or Graylog.

__init__(logger=<RootLogger root (WARNING)>, log_format='%(asctime)s %(levelname)s %(message)s', level=None)

capture(stdout_logger=<Logger STDOUT (WARNING)>, stderr_logger=<Logger STDERR (WARNING)>)

config_basic()

config_buffer()

config_file(file_name)

config_handler(handler)

config_stream(stdout_stream, stderr_stream=None)

class zensols.util.log.LogLevelSetFilter(levels)

Bases: object

__init__(levels)

filter(record)

class zensols.util.log.LoggerStream(logger, log_level=20)

Bases: object

Each line of standard out/error becomes a logged line

__init__(logger, log_level=20)

flush()

write(c)

class zensols.util.log.StreamLogDumper(stream, logger, level)

Bases: Thread

Redirect stream output to a logger in a running thread.

__init__(stream, logger, level)

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

static dump(stdout, stderr, logger)

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

zensols.util.log.add_logging_level(level_name, level_num, method_name=None)

Comprehensively adds a new logging level to the logging module and the currently configured logging class.

level_name becomes an attribute of the logging module with the value level_num. method_name becomes a convenience method for both logging itself and the class returned by logging.getLoggerClass() (usually just logging.Logger). If method_name is not specified, level_name.lower() is used.

To avoid accidental clobberings of existing attributes, this method will raise an AttributeError if the level name is already an attribute of the logging module or if the method name is already present.

See `Stack Overflow`_ for attribution.

zensols.util.log.add_trace_level()

Add a logging.TRACE logging level.

class zensols.util.log.loglevel(name='', level=10, init=None, enable=True)

Bases: object

Object used with a with scope that sets the logging level temporarily and sets it back.

Example:

with loglevel(__name__):
    logger.debug('test')

with loglevel(['zensols.persist', 'zensols.config'], init=True):
    logger.debug('test')

__init__(name='', level=10, init=None, enable=True)

Configure the temporary logging setup.

Parameters:

• name (Union[List[str], str, None]) – the name of the logger to set, or if a list is passed, configure all loggers in the list; if a string, configure all logger names split on spaces; if None or False, do not configure anything (handy for REPL prototyping); default to the root logger to log everything

• level (int) – the logging level, which defaults to logging.DEBUG

• init (Union[bool, int]) – if not None, initialize logging with logging.basicConfig() using the given level or True to use logging.WARNING

• enable (bool) – if False, disable any logging configuration changes for the block

zensols.util.package module

A convenience class around the pkg_resources module.

exception zensols.util.package.PackageError

Bases: APIError

Raised for errors related to packages from this module.

__annotations__ = {}

__module__ = 'zensols.util.package'

class zensols.util.package.PackageManager(pip_install_args=('--use-deprecated=legacy-resolver',), no_deps=False, force=False)

Bases: object

Gather and parse requirements and optionally install them.

__init__(pip_install_args=('--use-deprecated=legacy-resolver',), no_deps=False, force=False)

find_requirements(sources)

The requirements contained in this manager. .

Parameters:

sources (Tuple[Union[str, Path, PackageRequirement], ...]) – the :obj:PackageRequirement.spec`, requirements file, or directory with requirements files

Return type:

Tuple[PackageRequirement, ...]

force: bool = False

Whether to force installs and uninstalls regardless of whether a dependency is already (un)installed. This has no effect on the pip install command (--force).

get_installed_requirement(package)

Get an already installed requirement by name.

Parameters:

package (str) – the package name (i.e. zensols.util)

Return type:

Optional[PackageRequirement]

get_requirement(package)

First try to get an installed (get_installed_requirement), and if not found, back off to finding one with :class:().PackageResource`.

Parameters:

package (str) – the package name (i.e. zensols.util)

Return type:

Optional[PackageRequirement]

install(requirement)

Install a package in this Python enviornment with pip.

Parameters:

• requirement (PackageRequirement) – the requirement to install

• no_deps – if True do not install the package’s dependencies

Return type:

Optional[str]

Returns:

the output from the pip command invocation

is_installed(requirement)

Return whether a requirement is installed.

Return type:

bool

no_deps: bool = False

If True do not install the package’s dependencies.

pip_install_args: Tuple[str, ...] = ('--use-deprecated=legacy-resolver',)

Additional argument used for installing packages with pip.

uninstall(requirement)

Uninstall a package in this Python enviornment with pip.

Parameters:

• requirement (PackageRequirement) – the requirement to uninstall

• no_deps – if True do not install the package’s dependencies

Return type:

str

Returns:

the output from the pip command invocation

class zensols.util.package.PackageRequirement(requirement, source=None, meta=None)

Bases: Writable

A Python requirement specification.

__init__(requirement, source=None, meta=None)

property first_version: Version | None

The first specifier’s version if any specifiers exist.

classmethod from_spec(spec, **kwargs)

Return type:

Optional[PackageRequirement]

property is_strict: bool

Whether the requirement is strict around one and only one version.

meta: Dict[str, str] = None

The metadata of the package. This is populated by PackageManager.get_installed_requirement().

property name: str

The name of the module (i.e. zensols.someappname).

requirement: Requirement

The requirement of the package.

source: Path = None

The file in which the requirement was parsed.

property specifier: SpecifierSet

to_resource()

Create a package resource from this requirement.

Return type:

PackageRequirement

property url: str | None

The URL of the requirement.

property version: Version

The singleton version of the requirement.

Returns:

the version of the specifier or None if is_strict is False

class zensols.util.package.PackageResource(name)

Bases: Writable

Contains resources of installed Python packages. It makes the distribution available and provides access to to resource files with get_path() and as an index.

__init__(name)

property available: bool

Whether the package exists but not installed.

get_path(resource)

Return a resource file name by name. Optionally return resource as a relative path if the package does not exist.

Parameters:

resource (str) – a forward slash (/) delimited path (i.e. resources/app.conf) of the resource name

Return type:

Optional[Path]

Returns:

a path to that resource on the file system or None if the package doesn’t exist, the resource doesn’t exist

property installed: bool

Whether the package is installed.

name: str

The name of the module (i.e. zensols.someappname).

to_requirement()

The requirement represented by this instance.

Return type:

Optional[PackageRequirement]

property version: str | None

The version if the package is installed.

zensols.util.sci module

Scientific utilities.

class zensols.util.sci.ScientificUtil

Bases: object

A class containing utility methods.

static fixed_format(v, length=1, add_pad=False)

Format a number to a width resorting to scientific notation where necessary. The returned string is left padded with space in cases where scientific notation is too wide for v > 0. The mantissa is cut off also for v > 0 when the string version of the number is too wide.

Parameters:

length (int) – the length of the return string to include as padding

Return type:

str

zensols.util.std module

Utility classes for system based functionality.

class zensols.util.std.FileLikeType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Indicates readability aspects of a Python object. Enumeration values and helper methods specify how an object might be readable and in what way. This class describes types can be files-like, pathlib.Path, strings that point to files, and sys.stdin. The string or path STANDARD_OUT_PATH is treated as standard in.

filelike = 1

classmethod from_instance(instance)

Return an instance of this enumeration based on the type of instance.

Return type:

FileLikeType

property is_openable: bool

Whether an object can be opened with :function:`open`.

property is_readable: bool

Whether the enumeration value can be read.

static is_standard_in(instance)

Whether instance is not sys.stdin but represents it.

Return type:

bool

none = 5

path = 2

stdin = 4

string_path = 3

class zensols.util.std.openread(source, interpret_str=False, raise_error=True, no_close=False, *open_args, **open_kwargs)

Bases: object

Open an file object for reading if possible, and close it only if necessary. Input types are described by FileLikeType. If the source is

• Path: read a a file

• builtins.str: read as a file if interpret_str is True and it points to a file; otherwise, the string is read

• file-like object such as any IOBase (sub)class, it is

  read a file

Example:

import sys
import itertools as it

with openread(sys.stdin) as f:
    line = it.islice(f.readlines(), 1)

__init__(source, interpret_str=False, raise_error=True, no_close=False, *open_args, **open_kwargs)

Initialize.

Parameters:

• source (Any) – the data source

• interpret_str (bool) – whether to read the string or use it as a file name

• raise_error (bool) – if True raise OSError if source is an object that can not be opened or read

• no_close (bool) – do not close the file object (if opened in the first place), even if it should be

• open_kwargs – keyword argument passed to open()

class zensols.util.std.stdout(path=None, extension=None, recommend_name='unnamed', capture=False, logger=<Logger zensols.util.std (WARNING)>, open_args='w')

Bases: object

Write to a file or standard out. This is desigend to be used in command line application (CLI) applications with the zensols.cli module. Application class can pass a pathlib.Path to a method with this class.

Example:

def write(self, output_file: Path = Path('-')):
    """Write data.

    :param output_file: the output file name, ``-`` for standard out, or
                        ``+`` for a default

    """
    with stdout(output_file, recommend_name='unknown-file-name',
                extension='txt', capture=True, logger=logger):
        print('write data')

STANDARD_OUT_PATH: ClassVar[str] = '-'

The string used to indicate to write to standard out.

__init__(path=None, extension=None, recommend_name='unnamed', capture=False, logger=<Logger zensols.util.std (WARNING)>, open_args='w')

Initailize where to write. If the path is None or its name is STANDARD_OUT_PATH, then standard out is used instead of opening a file. If path is set to FILE_RECOMMEND_NAME, it is constructed from recommend_name. If no suffix (file extension) is provided for path then extesion is used if given.

Parameters:

• path (Union[str, Path]) – the path to write, or None

• extension (str) – the extension (sans leading dot .) to postpend to the path if one is not provied in the file name

• recommend_name (str) – the name to use as the prefix if path is not provided

• capture (bool) – whether to redirect standard out (sys.stdout) to the file provided by path if not already indicated to be standard out

• logger (Logger) – used to log the successful output of the file, which defaults to this module’s logger

• open_args (str) – the arguments given to open(), which defaults to w if none are given

classmethod is_stdout(path)

Return whether the path indicates to use to standard out.

Return type:

bool

class zensols.util.std.stdwrite(stdout=None, stderr=None)

Bases: object

Capture standard out/error.

__init__(stdout=None, stderr=None)

Initialize.

Parameters:

• stdout (TextIOBase) – the data sink for stdout (i.e. io.StringIO)

• stdout – the data sink for stderr

zensols.util.tempfile module

Classes to generate, track and clean up temporary files.

class zensols.util.tempfile.TemporaryFileName(directory=None, file_fmt='{name}', create=False, remove=True)

Bases: object

Create a temporary file names on the file system. Note that this does not create the file object itself, only the file names. In addition, the generated file names are tracked and allow for deletion. Specified directories are also created, which is only needed for non-system temporary directories.

The object is iterable, so usage with itertools.islice can be used to get as many temporary file names as desired. Calling instances (no arguments) generate a single file name.

__init__(directory=None, file_fmt='{name}', create=False, remove=True)

Initialize with file system data.

Parameters:

• directory (str) – the parent directory of the generated file

• file_fmt (str) – a file format that substitutes name for the create temporary file name; defaults to {name}

• create (bool) – if True, create directory if it doesn’t exist

• remove (bool) – if True, remove tracked generated file names if they exist

See tempfile:

clean()

Remove any files generated from this instance. Note this only deletes the files, not the parent directory (if it was created).

This does nothing if remove is False in the initializer.

property files: Tuple[Path, ...]

Return the file that have been created thus far.

class zensols.util.tempfile.tempfile(*args, **kwargs)

Bases: object

Generate a temporary file name and return the name in the with statement. Arguments to the form are the same as TemporaryFileName. The temporary file is deleted after completion (see TemporaryFileName).

Example:

with tempfile('./tmp', create=True) as fname:
    print(f'writing to {fname}')
    with open(fname, 'w') as f:
        f.write('this file will be deleted, but left with ./tmp\n')

__init__(*args, **kwargs)

zensols.util.time module

Peformance measure convenience utils.

class zensols.util.time.DurationFormatter(duration)

Bases: object

Utility class to format time as duration.

__init__(duration)

Initialize with the duration to format as seconds or a timedelta.

property days_hours_minutes_seconds: Tuple[int, int, int, int]

The days, hours, minutes and seconds of duration.

format(format)

Format the duration with using format, which is one of: :rtype: str

• hour_min_sec: HH:MM:SS

• non_zero: Hh:Mm:Ss with each field given if non_zero

exception zensols.util.time.TimeoutError

Bases: Exception

Raised when a time out even occurs in timeout() or timeprotect.

__module__ = 'zensols.util.time'

__weakref__

list of weak references to the object

class zensols.util.time.time(msg='finished', level=20, logger=None)

Bases: object

Used in a with scope that executes the body and logs the elapsed time.

Format f-strings are supported as the locals are taken from the calling frame on exit. This means you can do things like:

with time('processed {cnt} items'):
    cnt = 5
    tm.sleep(1)

which produeces: processed 5 items.

See the initializer documentation about special treatment for global loggers.

__init__(msg='finished', level=20, logger=None)

Create the time object.

If a logger is not given, it is taken from the calling frame’s global variable named logger. If this global doesn’t exit it logs to standard out. Otherwise, standard out/error can be used if given sys.stdout or sys.stderr.

Parameters:

• msg (str) – the message log when exiting the closure

• logger (Union[Logger, TextIOBase]) – the logger to use for logging or a file like object (i.e. sys.stdout) as a data sync

• level – the level at which the message is logged

static format_elapse(msg, seconds)

zensols.util.time.timeout(seconds=10, error_message='STREAM ioctl timeout')

This creates a decorator called @timeout that can be applied to any long running functions.

So, in your application code, you can use the decorator like so:

from timeout import timeout

# Timeout a long running function with the default expiry of
# TIMEOUT_DEFAULT seconds.
@timeout
def long_running_function1():
    pass

This was derived from the David Narayan’s `StackOverflow`_ thread.

class zensols.util.time.timeprotect(seconds=10, timeout_handler=None, context=None, error_message='STREAM ioctl timeout')

Bases: object

Invokes a block and bails if not completed in a specified number of seconds.

Parameters:

• seconds – the number of seconds to wait

• timeout_handler – function that takes a single argument, which is this timeprotect object instance; if None, then nothing is done if the block times out

• context – an object accessible from the timeout_hander via self, which defaults to None

See:

timeout()

__init__(seconds=10, timeout_handler=None, context=None, error_message='STREAM ioctl timeout')

zensols.util.warn module

Utilities to silence warnings.

class zensols.util.warn.WarningSilencer(filters)

Bases: object

A utility class to invoke the Python warnings system to suppress (ignore) warnings. Both the category (warning class) and a regular expression can be provided.

__init__(filters)

property filter_parameters: Tuple[Dict[str, Any], ...]

A list of dictionaries, each of which are used as keyword parameters to :function:`warning.filterwarnings`.

filters: Tuple[Dict[str, str], ...]

A tuple of dictionaries that will be given to the :function:`warning.filterwarnings` function. This is meant to be added to an application configuration.

For example, in a Zensols YAML configuration:

warn_silencer:
    class_name: zensols.util.warn.WarningSilencer
    filters:
      - action: 'ignore'
        category: FutureWarning
        message: '^Some error message.*'

zensols.util.writable module

A class that allows human readable information (sometimes debugging) output with a hierarchical structure.

class zensols.util.writable.Writable

Bases: object

An interface for classes that have multi-line debuging capability.

_write(c)

Use WritableContext as a data sink.

classmethod _trunc(s, max_len=None)

Return type:

str

_sp(depth)

Utility method to create a space string.

Return type:

str

_set_indent(indent=None)

Set the indentation for the instance. By default, this value is 4.

Parameters:

indent (int) – the value to set as the indent for this instance, or None to unset it

_write_line(line, depth, writer, max_len=False, repl_newlines=False)

Write a line of text line with the correct indentation per depth to writer.

Parameters:

max_line – truncate to the given length if an int or WRITABLE_MAX_COL if True

Repl_newlines:

whether to replace newlines with spaces

_write_block(lines, depth, writer, limit=None)

Write a block of text with indentation.

Parameters:

limit (int) – the max number of lines in the block to write

_write_wrap(text, depth, writer, width=None, **kwargs)

Like _write_line() but wrap text per width.

Parameters:

• text (str) – the text to word wrap

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

• width (int) – the width of the text before wrapping, which defaults to WRITABLE_MAX_COL

• kwargs – the keyword arguments given to textwarp.wrap()

_write_object(obj, depth, writer)

Write an object based on the class of the instance.

_write_iterable(data, depth, writer, include_index=None)

Write list data with the correct indentation per depth to writer.

Parameters:

include_index (bool) – if True, add an incrementing index for each element in the output

_write_dict(data, depth, writer, inline=False, one_line=False)

Write dictionary data with the correct indentation per depth to writer.

Parameters:

• data (Dict) – the data wto write

• inline (bool) – whether to write values in one line (separate from key)

• one_line (bool) – whether to print all of data on one line

WRITABLE_INCLUDE_INDEX: ClassVar[bool] = False

Whether to include index numbers with levels in sequences.

WRITABLE_INDENT_SPACE: ClassVar[int] = 4

The default number of spaces to indent each level.

WRITABLE_MAX_COL: ClassVar[int] = 80

The default maximum column size before wrapping text.

write(depth=0, writer=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write the contents of this instance to writer using indention depth.

Parameters:

• depth (int) – the starting indentation depth

• writer (TextIOBase) – the writer to dump the content of this writable

write_to_log(logger, level=20, depth=0, split_lines=True)

Just like write() but write the content to a log message.

Parameters:

• logger (Logger) – the logger to write the message containing content of this writable

• level (int) – the logging level given in the logging module

• depth (int) – the starting indentation depth

• split_lines (bool) – if True each line is logged separately

class zensols.util.writable.WritableContext(target, depth, writer)

Bases: object

A text data sync given to a Writable as a convenience object.

__init__(target, depth, writer)

depth: int

The text indentation.

target: Writable

The client of this class.

writer: TextIOBase

The data sync to which text gets written.

Module contents