honeycomb.utils package

Submodules

honeycomb.utils.config_utils module

Honeycomb Config Utilities.

honeycomb.utils.config_utils.config_field_type(field, cls)[source]

Validate a config field against a type.

Similar functionality to validate_field_matches_type() but returns honeycomb.defs.ConfigField

honeycomb.utils.config_utils.get_config_parameters(plugin_path)[source]

Return the parameters section from config.json.

honeycomb.utils.config_utils.get_truetype(value)[source]

Convert a string to a pythonized parameter.

honeycomb.utils.config_utils.is_valid_field_name(value)[source]

Ensure field name is valid.

honeycomb.utils.config_utils.process_config(ctx, configfile)[source]

Process a yaml config with instructions.

This is a heavy method that loads lots of content, so we only run the imports if its called.

honeycomb.utils.config_utils.validate_config(config_json, fields)[source]

Validate a JSON file configuration against list of honeycomb.defs.ConfigField.

honeycomb.utils.config_utils.validate_config_parameters(config_json, allowed_keys, allowed_types)[source]

Validate parameters in config file.

honeycomb.utils.config_utils.validate_field(field, allowed_keys, allowed_types)[source]

Validate field is allowed and valid.

honeycomb.utils.config_utils.validate_field_matches_type(field, value, field_type, select_items=None, _min=None, _max=None)[source]

Validate a config field against a specific type.

honeycomb.utils.daemon module

Honeycomb DaemonRunner utility.

class honeycomb.utils.daemon.myRunner(app, pidfile=None, stdout=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, stderr=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, stdin=<_io.TextIOWrapper name='/dev/null' mode='rt' encoding='UTF-8'>)[source]

Bases: daemon.runner.DaemonRunner

Overriding default runner behaviour to be simpler.

Override init to fit honeycomb needs.

We initialize app with default stdout/stderr from sys instead of file path and remove the use of parse_args() since it’s not actually a standalone runner

honeycomb.utils.plugin_utils module

Honeycomb generic plugin install utils.

exception honeycomb.utils.plugin_utils.CTError(errors)[source]

Bases: Exception

Copytree exception class, used to collect errors from the recursive copy_tree function.

Collect errors.

Parameters:errors – Collected errors
honeycomb.utils.plugin_utils.copy_file(src, dst)[source]

Copy a single file.

Parameters:
  • src – Source name
  • dst – Destination name
honeycomb.utils.plugin_utils.copy_tree(src, dst, symlinks=False, ignore=[])[source]

Copy a full directory structure.

Parameters:
  • src – Source path
  • dst – Destination path
  • symlinks – Copy symlinks
  • ignore – Subdirs/filenames to ignore
honeycomb.utils.plugin_utils.get_plugin_path(home, plugin_type, plugin_name, editable=False)[source]

Return path to plugin.

Parameters:
  • home – Path to honeycomb home
  • plugin_type – Type of plugin (honeycomb.defs.SERVICES pr honeycomb.defs.INTEGRATIONS)
  • plugin_name – Name of plugin
  • editable – Use plugin_name as direct path instead of loading from honeycomb home folder
honeycomb.utils.plugin_utils.get_select_items(items)[source]

Return list of possible select items.

honeycomb.utils.plugin_utils.install_deps(pkgpath)[source]

Install plugin dependencies using pip.

We import pip here to reduce load time for when its not needed.

honeycomb.utils.plugin_utils.install_dir(pkgpath, install_path, register_func, delete_after_install=False)[source]

Install plugin from specified directory.

install_path and register_func are same as install_plugin(). :param delete_after_install: Delete pkgpath after install (used in install_from_zip()).

honeycomb.utils.plugin_utils.install_from_repo(pkgname, plugin_type, install_path, register_func)[source]

Install plugin from online repo.

honeycomb.utils.plugin_utils.install_from_zip(pkgpath, install_path, register_func, delete_after_install=False)[source]

Install plugin from zipfile.

honeycomb.utils.plugin_utils.install_plugin(pkgpath, plugin_type, install_path, register_func)[source]

Install specified plugin.

Parameters:
  • pkgpath – Name of plugin to be downloaded from online repo or path to plugin folder or zip file.
  • install_path – Path where plugin will be installed.
  • register_func – Method used to register and validate plugin.
honeycomb.utils.plugin_utils.list_local_plugins(plugin_type, plugins_path, plugin_details)[source]

List local plugins with details.

honeycomb.utils.plugin_utils.list_remote_plugins(installed_plugins, plugin_type)[source]

List remote plugins from online repo.

honeycomb.utils.plugin_utils.parse_plugin_args(command_args, config_args)[source]

Parse command line arguments based on the plugin’s parameters config.

Parameters:
  • command_args – Command line arguments as provided by the user in key=value format.
  • config_args – Plugin parameters parsed from config.json.
Returns:

Validated dictionary of parameters that will be passed to plugin class

honeycomb.utils.plugin_utils.print_plugin_args(plugin_path)[source]

Print plugin parameters table.

honeycomb.utils.plugin_utils.uninstall_plugin(pkgpath, force)[source]

Uninstall a plugin.

Parameters:
  • pkgpath – Path to package to uninstall (delete)
  • force – Force uninstall without asking

honeycomb.utils.tailer module

Honeycomb service log tailer.

class honeycomb.utils.tailer.Tailer(name: str, filepath: str, color: str = '', nlines: int = 10, follow: bool = False, outfile=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, sleeptime: int = 0.5, show_name: bool = True, used_colors: list = [])[source]

Bases: object

Colorized file tailer.

Print lines from a file prefixed with a colored name. Optionally continue to follow file.

follow_file()[source]

Follow a file and send every new line to a callback.

print_log(line)[source]

Print a line from a logfile.

print_named_log(line)[source]

Print a line from a logfile prefixed with service name.

stop()[source]

Stop follow.

honeycomb.utils.wait module

Honeycomb wait utilities.

exception honeycomb.utils.wait.TimeoutException[source]

Bases: Exception

Exception to be raised on timeout.

honeycomb.utils.wait.search_json_log(filepath, key, value)[source]

Search json log file for a key=value pair.

Parameters:
  • filepath – Valid path to a json file
  • key – key to match
  • value – value to match
Returns:

First matching line in json log file, parsed by json.loads()

honeycomb.utils.wait.wait_until(func, check_return_value=True, total_timeout=60, interval=0.5, exc_list=None, error_message='', *args, **kwargs)[source]

Run a command in a loop until desired result or timeout occurs.

Parameters:
  • func – Function to call and wait for
  • check_return_value (bool) – Examine return value
  • total_timeout (int) – Wait timeout,
  • interval (float) – Sleep interval between retries
  • exc_list (list) – Acceptable exception list
  • error_message (str) – Default error messages
  • args – args to pass to func
  • kwargs – lwargs to pass to fun

Module contents

Honeycomb Utils.