zensols.util package

Submodules

zensols.util.executor

Simplifies external process handling.

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

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 (if defined)

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

Bases: object

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

Utilities for hashing text.

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

Bases: object

__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

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.pkgres

A convenience class around the pkg_resources module.

class zensols.util.pkgres.PackageResource(name, file_system_defer=True)

Bases: object

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, file_system_defer=True)

property distribution: DistInfoDistribution | None

The package distribution.

Returns:

the distribution or None if it is not installed

property exists: bool

Return if the package exists and installed.

file_system_defer: bool = True

Whether or not to return resource paths that point to the file system when this package distribution does not exist.

See:

get_path()

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 and file_system_defer is False

name: str

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

property version: str | None

Return the version if the package exists.

zensols.util.sci

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

Utility classes for system based functionality.

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')

FILE_RECOMMEND_NAME: ClassVar[str] = '+'

The string used to indicate to use the recommended file name.

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_file_recommend(path)

Return whether the path indicates to recommmend a file.

Return type:

bool

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

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

Peformance measure convenience utils.

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 (if defined)

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')

Module contents