toolrack.aio

Utilities based on the asyncio library.

exception toolrack.aio.AlreadyRunning

The TimedCall is already running.

exception toolrack.aio.NotRunning

The TimedCall is not running.

class toolrack.aio.PeriodicCall(func: Callable, *args, **kwargs)

A TimedCall called at a fixed time intervals.

start(interval: Union[int, float], now: bool = True)

Start calling the function periodically.

Parameters:

• interval – the time interval in seconds between calls.
• now – whether to make the first call immediately.

class toolrack.aio.ProcessParserProtocol(future: _asyncio.Future, out_parser=None, err_parser=None)

Collect process stdout and stderr.

Line parser functions can be passed for stdout and stderr, and they are called on each full line of output.

When the process ends, the future returns a tuple with the full stdout and stderr. Each tuple element is None if a parser is passed for that stream.

Parameters:

• future (asyncio.Future) – a Future called with a tuple with (stdout, stderr) from the process once it it exits.
• out_parser (callable) – an optional parser for the process standard output.
• err_parser (callable) – an optional parser for the process standard error.

connection_lost(exc)

Called when the connection is lost or closed.

The argument is an exception object or None (the latter meaning a regular EOF is received or the connection was aborted or closed).

pipe_data_received(fd, data)

Called when the subprocess writes data into stdout/stderr pipe.

fd is int file descriptor. data is bytes object.

class toolrack.aio.StreamHelper(callback: Optional[Callable[[str], None]] = None, separator: str = 'n')

Helper to cache data until full lines of text are received.

This is useful to collect data from a stream and process them when full lines are received. For example:

stream = StreamHelper(callback)
stream.receive_data('line one\nline two')
stream.receive_data('continues here\n')

would call callback twice, one with 'line one' and one with 'line two continues here'

Parameters:

• callback (callable) – an optional function which is called with full lines of text from the stream.
• separator (str) – the line separator

flush_partial()

Flush and process pending data from a partial line.

get_data()

Return the full content of the stream.

receive_data(data: str)

Receive data and process them.

If a callback has been passed to the class, it’s called for each full line of text.

class toolrack.aio.TimedCall(func: Callable, *args, **kwargs)

Call a function based on a timer.

The class takes a function with optional arguments. Upon start(), the function is scheduled at specified times until stop() is called (or the time iterator is exausted).

Parameters:

• func – the function to call periodically.
• args – arguments to pass to the function.
• kwargs – keyword arguments to pass to the function.

running

Whether the PeriodicCall is currently running.

start(times_iter: Iterable[Union[float, int]])

Start calling the function at specified times.

Parameters:times_iter – an iterable yielding times to execute the function at. If the iterator exhausts, the TimedCall is stopped. Times must be compatible with loop.time().

stop()

Stop calling the function periodically.

It returns an asyncio.Future to wait for the stop to complete.

toolrack.certificate

Utility functions for SSL/TSL certificate handling.

toolrack.certificate.get_host_certificate(uri: str, get_func: Callable[[Tuple[str, int]], str] = <function get_server_certificate>) → str

Return a string with the host certificate.

Parameters:uri (str) – the host URI, in the form [scheme://]host[:port]. The scheme is optional (and ignored), and port defaults to 443.

toolrack.collect

Collection of objects of the same kind.

A Collection holds objects identifying them by the value of an attribute. For instance:

collection = Collection('SomeObject', 'name')
collection.add(obj)

will use obj.name as key and:

collection.get('foo')

will return the object with obj.name == foo.

The Collection is iterable, and yields the contained objects:

for obj in collection:
    # ... do something with obj

class toolrack.collect.Collection(obj_type: type, key: str)

A Collection of objects keyed on an attribute.

It collects objects identified by the value of an attribute. No objects with duplicated keys are allowed.

Parameters:

• obj_type – string identifying the objects type.
• key – the object attribute to use as key.

add(obj: Any)

Add and return an object.

clear()

Empty the collection.

get(key: str) → Any

Return the object with the specified key.

keys() → Iterator[str]

Return an iterator with collection keys.

remove(key: str) → Any

Remove and return the object with the specified key.

sorted() → List[T]

Return a list of objects sorted by key.

exception toolrack.collect.DuplicatedObject(obj_type: type, obj_key: str)

An object with the specified key is already the Collection.

exception toolrack.collect.UnknownObject(obj_type, obj_key: str)

No object with the specified key in the Collection.

toolrack.config

Hold and parse key/value configurations.

The Config describes valid configuration keys along with their types, and performs parsing and type conversion from a dict of config options. It also checks required keys and applies default values if an option is not provided.

As an example:

config = Config(
    ConfigKey('option1', 'int', default=4),
    ConfigKey('option2', 'bool', required=True))
config.parse({'option2': 'true'})

returns {'option1': 4, 'option2': True}.

class toolrack.config.Config(*keys)

Parse a configuration dictionary.

A configuration has a set of keys of specific types.

extend(*keys)

Return a new Config with additional keys.

keys()

Return ConfigKeys sorted by name alphabetically.

parse(config: Optional[Dict[str, Any]]) → Dict[str, Any]

Parse the provided configuration dict.

Returns a dict with configuration keys and values converted to the proper type. The dict includes only keys declared in the Config, with default values if not present in the config dict.

class toolrack.config.ConfigKey(name: str, _type: str, description: str = '', required: bool = False, default: Optional[Any] = None, validator: Optional[Callable[[Any], None]] = None)

A key in the Configuration.

parse(value)

Convert and validate a value.

validate(value)

Validate a value based for the key.

Can be overridden by subclasses. It should raise a ValueError if the value is invalid.

class toolrack.config.ConfigKeyTypes

Collection of type converters for ConfigKeys.

get_converter(_type: str) → Callable

Return the converter method for the specified type.

exception toolrack.config.InvalidConfigValue(key: str)

exception toolrack.config.MissingConfigKey(key: str)

toolrack.convert

Utilities for unit conversion.

toolrack.convert.BYTE_SUFFIXES = (None, 'kib', 'mib', 'gib', 'tib', 'pib', 'eib', 'zib', 'yib')

Binary byte multipliers

toolrack.convert.convert_bbyte(value, suffix=None, to=None)

Convert a binary byte value across multipliers.

Parameters:

• value (int) – the current value.
• suffix (str) – the current multiplier for the value (None for bytes).
• to (str) – the target multiplier (None for bytes).

toolrack.fsmap

Access the filesystem in a dict-like fashion.

This module provides a Directory class which provides access to the filesystem subtree below its path, allow accessing files and sub-directories as elements of a dict (e.g. directory['foo'] or directory['foo/bar']).

toolrack.fsmap.DIR = <object object>

Marker for creating directories.

class toolrack.fsmap.Directory(path)

Provide access to the sub-tree of a directrory.

It represents a directory in the filesystem:

directory = Directory('/base/path')

The object is iterable and yields names of contained elements:

for elem in directory:
    do_something(directory[elem])

Sub-directories and files below the base path can be accessed as items of a dict. For instance:

directory['a-dir']['a-file']

or even with a single access, using OS path format:

directory['a-dir/a-file']

Path elements can be removed with del:

del directory['a-file']
del directory['a-dir']  # this will delete the whole sub-tree

Files are created/overwritten by assiging content:

directory['a-file'] = 'some content'

and directories are created using the DIR marker:

directory['a-new-dir'] = DIR

toolrack.iterate

Utility functions for iterables.

toolrack.iterate.flatten_dict(data, join_char='.', prefix='')

Flatten a nested dict to (key, value) tuples.

A neted dict like:

{'foo': {'bar': 3, 'baz': 4},
 'bza': 'something'}

is flattened in a sequence of tuples like:

('foo.bar': 3), ('foo.baz': 4), ('bza': 'something')

Parameters:

• data (dict) – a dict to flatten.
• join_char (str) – the character to use to join key tokens.
• prefix (str) – an optional prefix to prepend to keys.

toolrack.json_util

Utilities for dealing with JSON data.

toolrack.json_util.indent(in_fd, out_fd, indent=4, ensure_ascii=False)

Indent JSON data.

It reads text in JSON format from in_fd and writes the formatted output to out_fd, using the specified amount of indent spaces.

Parameters:

• in_fd – input file descriptor.
• out_fd – output file descriptor.
• indent (int) – number of spaces used for indentation.
• ensure_ascii (bool) – passed to the JSON serializer, if specified, non-ASCII characters are escaped.

toolrack.log

Logging helpers.

The Loggable mixin provides a logger attribute with a configured logger using the class name (including the module path) as logger name.

toolrack.log.LOG_FORMAT = '%(asctime)s - %(levelname)s - %(name)s - %(message)s'

Default format for logging to a stream

class toolrack.log.Loggable

Mixin class providing a logger attribute.

logger

Decorator to a class method a cached property.

The property method is called just the first time for an instance, and its result cached.

toolrack.log.setup_logger(name=None, stream=None, level=None, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')

Helper to setup a logger with a default handler.

Parameters:

• name (str) – the name of the logger. If not specified, the root logger is used.
• stream – an output stream for the log handler. If not specified, the null handler is installed.
• level – the minimum log level for the logger.

toolrack.password

Password generation functions.

PasswordProfile defines a set of characters to use for generating password. It’s creating by passing a string with characters or character definitions encosed in curly braces (such as {alnum}, {num}, {alpha}), which are expanded to the corresponding set of characters.

For instance:

profile = PasswordProfile('{alpha}-_')
profile.generate(length=5)

yields a 5-chars password composed of letters, dashes and underscores.

toolrack.password.DEFAULT_CHARS = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!"#$%&\'()*+,-./:;<=>?@[\\]^_`{|}~'

Default character set: letters, numbers and punctuation

toolrack.password.DEFAULT_LENGTH = 10

Default password length

class toolrack.password.PasswordProfile(definition)

A password profile, specifying how to generate a random password.

CHAR_DEFS = {'alnum': 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789', 'alpha': 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ', 'num': '0123456789', 'punct': '!"#$%&\'()*+,-./:;<=>?@[\\]^_`{|}~', 'space': ' \t\n\r\x0b\x0c'}

chars

Return the set of characters used in generation.

generate(length=10)

Generate a random password.

toolrack.password.generate_password(chars='abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!"#$%&\'()*+, -./:;<=>?@[\\]^_`{|}~', length=10)

Generate a random password using the supplied characters.

Parameters:

• chars (str) – a string with chars to choose from.
• length (int) – number of chars for the password.

toolrack.path

Functions for paths handling.

toolrack.path.match_files(dirpaths, patterns, ignorecase=False)

Search files by name based on shell patterns.

Parameters:

• dirpaths (list) – a list of paths to search from.
• patterns (list) – a list of name patterns to match.

Returns:

an iterator yielding matched files.

toolrack.property

Property decorators.

class toolrack.property.cachedproperty(func)

Decorator to a class method a cached property.

The property method is called just the first time for an instance, and its result cached.

toolrack.script

Base class for python scripts.

This module provides a Script class to reduce boilerplate when creating python scripts.

Script instances are callable, and can optionally receive a list of arguments (by default they look at sys.argv).

A typical use of Script is to declare a subclass with the script logic and create an instance:

class MyScript(Script):

    def main(self, args):
       # script logic here
       ...

my_script = MyScript()

The instance can be referenced in setuptools entry_points key:

setup(
    entry_points={'console_scripts': ['my_script=path.to.script:my_script']},
    ...)

exception toolrack.script.ErrorExitMessage(message: str, code: int = 1)

Raised to exit the process with the specified message and exit code.

Parameters:

• message – the error message.
• code – the script exit code.

class toolrack.script.Script(stdout: Optional[IO] = None, stderr: Optional[IO] = None)

Wraps a python script handling argument parsing.

Subclasses must implement get_parser() and main() methods.

Inside main(), ErrorExitMessage can be raised with the appropriate message and code to cause the script termination, with the message outputted to standard error.

Script instances are callable, and can be passed the argument list (which defaults to sys.argv if not provided).

exit(code: int = 0)

Exit with the specified return code.

get_parser() → argparse.ArgumentParser

Return a configured argparse.ArgumentParser instance.

Note

Subclasses must implement this method.

handle_keyboard_interrupt(interrupt: KeyboardInterrupt)

Called when a KeyboardInterrupt is raised.

By default it just traps the exception and exits with success. It can be overridden to perform additional cleanups.

main(args: argparse.Namespace)

The body of the script.

It gets called with the argparse.Namespace instance returned by get_parser().

Parameters:args – command line arguments.

Note

Subclasses must implement this method.

toolrack.testing

Unit-test features.

class toolrack.testing.Dir(path: pathlib.Path)

A helper for creating files and directories under a base directory.

This is meant to be used for test fixtures.

join(*paths)

Join the specified path fragments with directory prefix.

mkdir(path=None)

Create a temporary directory and return the pathlib.Path.

By default, a random name is chosen.

Parameters:path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().

mkfile(path=None, content='', mode=None)

Create a temporary file and return the pathlib.Path.

By default, a random name is chosen.

Parameters:

• path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().
• content (str) – the content of the file.
• mode (int) – Unix permissions for the file.

mksymlink(target, path=None)

Create a symbolic link and return the pathlib.Path.

By default, a random name is chosen.

Parameters:

• target – path of the symlink target.
• path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().

toolrack.testing.tempdir(tmpdir)

A temporary directory fixture.

toolrack.testing.fixtures

Unit-test fixtures.

class toolrack.testing.fixtures.Dir(path: pathlib.Path)

A helper for creating files and directories under a base directory.

This is meant to be used for test fixtures.

join(*paths)

Join the specified path fragments with directory prefix.

mkdir(path=None)

Create a temporary directory and return the pathlib.Path.

By default, a random name is chosen.

Parameters:path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().

mkfile(path=None, content='', mode=None)

Create a temporary file and return the pathlib.Path.

By default, a random name is chosen.

Parameters:

• path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().
• content (str) – the content of the file.
• mode (int) – Unix permissions for the file.

mksymlink(target, path=None)

Create a symbolic link and return the pathlib.Path.

By default, a random name is chosen.

Parameters:

• target – path of the symlink target.
• path – if specified, it’s appended to the base directory and all intermiediate directories are created too. A relative path must be specified. A tuple of strings can be also passed, in which case elements are joined using os.path.sep().

toolrack.testing.fixtures.tempdir(tmpdir)

A temporary directory fixture.

toolrack.threading

Thread-related utilities.

class toolrack.threading.ThreadLocalAttribute(name)

Descriptor to proxy access to a class attribute, making it thread-local.

This descriptor can be used to make a class attribute thread-local in a trasparent way:

class MyClass:

    attr = ThreadLocalAttribute('attr')


instance = MyClass()

The attribute will be normally accessible as instance.attr, but it’s stored in a threading.local() context.

toolrack.threading.thread_local_attrs(*attrs)

Class decorator to make attributes storage thread-local.

It should be passed names of attributes in the decorated class to make local:

@thread_local_attrs('foo', 'bar')
class MyClass:

    foo = 3
    bar = None

ToolRack

A collection of utility functions and classes, and a few scripts too.

Documentation

API docs are available on ReadTheDocs.

Installation

ToolRack can be installed from PyPI.

As a user run:

$ pip install toolrack

Development installation

The source tree is available available at https://github.com/albertodonato/toolrack.

Development setup can be done via Virtualenv.

As a user run:

$ python3 -m venv <target-dir>
$ . <target-dir>/bin/activate
$ git clone https://github.com/albertodonato/toolrack
$ cd toolrack
$ python3 setup.py develop

Indices and tables

• Index
• Module Index
• Search Page