Helpers¶

Config¶

Command-line configuration parsing and client builder helper functions

es_client.helpers.config.cli_opts(value: str, settings: Dict | None = None, onoff: Dict | None = None, override: Dict | None = None) → Tuple[Tuple[str], Dict]¶

Parameters:

value (str) – The command-line option name. The key must be present in settings, or in CLICK_SETTINGS
settings (dict) – A dictionary consisting of click.Option names as keys, with each key having a dictionary consisting of click.Option parameter names as keys, with their associated settings as the value. If settings is not provided, it will be populated by CLICK_SETTINGS.
onoff (dict) – A dictionary consisting of the keys on and off, with values used to set up a Click boolean option, .e.g. {'on': '', 'off': 'no-'}. See below for examples.
override (dict) – A dictionary consisting of keys in settings with values you wish to override.

Return type:

Tuple

Returns:

A value suitable to use with the click.option() decorator, appearing as a tuple containing a tuple and a dictionary, e.g.

(('--OPTION1',),{'key1', 'value1', ...})

Click uses decorators to establish options and arguments for a command. The parameters specified for these decorator functions can be stored as default dictionaries, then expanded and overridden, if desired.

In the cli_example.py file, the regular click.option decorator function is wrapped by option_wrapper(), and is aliased as click_opt_wrap. This wrapped decorator in turn calls this function and utilizes * arg expansion. If settings is None, default values from CLICK_SETTINGS, are used to populate settings. This function calls override_settings() to override keys in settings with values from matching keys in override.

In the example file, this looks like this:

import click
from es_client.helpers.utils import option_wrapper
defaults.ONOFF = {'on': '', 'off': 'no-'}
click_opt_wrap = option_wrapper()
# ...
@click.group(context_settings=context_settings())
@click_opt_wrap(*cli_opts('OPTION1', settings={KEY: NEWVALUE}))
@click_opt_wrap(*cli_opts('OPTION2', onoff=tgl))
# ...
@click_opt_wrap(*cli_opts('OPTIONX'))
@click.pass_context
def run(ctx, OPTION1, OPTION2, ..., OPTIONX):
    # code here

The default setting KEY of OPTION1 would be overriden by NEWVALUE.

OPTION2 automatically becomes a Click boolean option, which splits the option into an enabled/disabled dichotomy by option name. In this example, it will be rendered as:

'--OPTION2/--no-OPTION2'

The dictionary structure of defaults.ONOFF is what this what this function requires, i.e. an on key and an off key. The values for on and off can be whatever you like, e.g.

defaults.ONOFF = {'on': 'enable-', 'off': 'disable-'}

which, based on the above example, would render as:

'--enable-OPTION2/--disable-OPTION2'

It could also be:

defaults.ONOFF = {'on': 'monty-', 'off': 'python-'}

which would render as:

'--monty-OPTION2/--python-OPTION2'

but that would be too silly.

A ConfigurationError is raised value is not found as a key in settings, or if the onoff parsing fails.

es_client.helpers.config.cloud_id_override(args: Dict, ctx: Context) → Dict¶

Parameters:

args (dict) – A dictionary built from ctx.params keys and values.
ctx (Context) – The Click command context

Return type:

dict

Returns:

Updated version of args

If hosts are defined in the YAML configuration file, but cloud_id is specified at the command-line, we need to remove the hosts parameter from the configuration dictionary built from the YAML file before merging. Command-line provided arguments always supersede configuration file ones. In this case, cloud_id and hosts are mutually exclusive, and the command-line provided cloud_id must supersede a configuration file provided hosts.

This function returns an updated dictionary args to be used for the final configuration as well as updates the ctx.obj['client_args'] object. It’s simply easier to merge dictionaries using a separate object. It would be a pain and unnecessary to make another entry in ctx.obj for this.

es_client.helpers.config.context_settings() → Dict¶

Return type:: dict
Returns:: kwargs suitable to be used as Click Command context_settings parameters.

Includes the terminal width from get_width()

Help format settings:

help_option_names=['-h', '--help']

The default context object (ctx.obj) dictionary:

obj={'default_config': None}

And automatic environment variable reading based on a prefix value:

auto_envvar_prefix=ENV_VAR_PREFIX

from ENV_VAR_PREFIX

es_client.helpers.config.generate_configdict(ctx: Context) → None¶

Parameters:: ctx (Context) – The Click command context
Return type:: None

Generate a client configuration dictionary from ctx.params and ctx.obj['default_config'] (if provided), suitable for use as the VALUE in Builder(configdict=VALUE)

It is stored as ctx.obj['default_config'] and can be referenced after this function returns.

The flow of this function is as follows:

Step 1: Call get_arg_objects() to create ctx.obj['client_args'] and ctx.obj['other_args'], then update their values from ctx.obj['draftcfg'] (which was populated by get_config()).

Step 2: Call override_client_args() and override_other_args(), which will use command-line args from ctx.params to override any values from the YAML configuration file.

Step 3: Populate ctx.obj['configdict'] from the resulting values.

es_client.helpers.config.get_arg_objects(ctx: Context) → None¶

Parameters:: ctx (Context) – The Click command context
Return type:: None

Set ctx.obj['client_args'] as a DotMap object, and ctx.obj['other_args'] as an DotMap object.

These will be updated with values returned from check_config(ctx.obj['draftcfg']).

ctx.obj['draftcfg'] was populated when get_config() was called.

es_client.helpers.config.get_client(configdict: Dict | None = None, configfile: str | None = None, autoconnect: bool = False) → Elasticsearch¶

Parameters:

configdict – A configuration dictionary
configfile – A YAML configuration file
autoconnect – Connect to client automatically

Returns:

A client connection object

Return type:

Elasticsearch

Get an Elasticsearch Client using Builder

Build a client connection object out of settings from configfile or configdict.

If neither configfile nor configdict is provided, empty defaults will be used.

If both are provided, configdict will be used, and configfile ignored.

Raises ESClientException if unable to connect.

es_client.helpers.config.get_config(ctx: Context, quiet: bool = True) → Context¶

Parameters:

ctx (Context) – The Click command context
quiet (bool) – If the default configuration file is being used, suppress the STDOUT message indicating that.

Return type:

None

If ctx.params['config'] is a valid path, return the validated dictionary from the YAML.

If nothing has been provided to ctx.params['config'], but ctx.obj['default_config'] is populated, use that, and write a line to STDOUT explaining this, unless quiet is True.

Writing directly to STDOUT is done here because logging has not yet been configured, nor can it be as the configuration options are just barely being read.

Store the result in ctx.obj['draftcfg']

es_client.helpers.config.get_hosts(ctx: Context) → Sequence[str] | None¶

Parameters:: ctx (Context) – The Click command context
Returns:: A list of hosts
Return type:: list

Return a list of hosts suitable for ClientArgs.hosts from ctx.params['hosts'], validating the url schema for Elasticsearch compliance for each host provided.

Raises a ConfigurationError if schema validation fails.

es_client.helpers.config.get_width() → Dict¶

Return type:: dict
Returns:: A dictionary suitable for use by itself as the Click Command context_settings parameter.

Determine terminal width by calling shutil.get_terminal_size()

Return value takes the form of {"max_content_width": get_terminal_size()[0]}

es_client.helpers.config.hosts_override(args: Dict, ctx: Context) → Dict¶

Parameters:

args (dict) – A dictionary built from ctx.params keys and values.
ctx (Context) – The Click command context

Return type:

dict

Returns:

Updated version of args

If hosts are provided at the command-line and are present in ctx.params['hosts'], but cloud_id was in the config file, we need to remove the cloud_id key from the configuration dictionary built from the YAML file before merging. Command-line provided arguments always supersede configuration file ones, including hosts overriding a file-based cloud_id.

This function returns an updated dictionary args to be used for the final configuration as well as updates the ctx.obj['client_args'] object. It’s simply easier to merge dictionaries using a separate object. It would be a pain and unnecessary to make another entry in ctx.obj for this.

es_client.helpers.config.options_from_dict(options_dict) → Callable¶: Build Click options decorators programmatically

es_client.helpers.config.override_client_args(ctx: Context) → None¶

Parameters:: ctx (Context) – The Click command context
Return type:: None

Override ctx.obj['client_args'] settings with any: values found in ctx.params

Update ctx.obj['client_args'] with the results.

In the event that there are neither hosts nor a cloud_id after the updates, log to debug that this is the case, and that the default value for hosts of http://127.0.0.1:9200 will be used.

es_client.helpers.config.override_other_args(ctx: Context) → None¶

Parameters:: ctx (Context) – The Click command context
Return type:: None

Override ctx.obj['other_args'] settings with any values found in ctx.params

Update ctx.obj['other_args'] with the results.

es_client.helpers.config.override_settings(settings: Dict, override: Dict) → Dict¶

Parameters:

settings (dict) – The source data
override (dict) – The data which will override settings

Return type:

dict

Returns:

An dictionary based on settings updated with values from override

This function is called by cli_opts() in order to override settings used in a Click Option.

Click uses decorators to establish options and arguments for a command. The parameters specified for these decorator functions can be stored as default dictionaries, then expanded and overridden, if desired.

In the cli_example.py file, the regular click.option decorator function is wrapped by option_wrapper(), and is aliased as click_opt_wrap. This wrapped decorator in turn calls cli_opts() and utilizes * arg expansion. cli_opts() references defaults, and calls this function to override keys in settings with values from matching keys in override.

In the example file, this looks like this:

import click
from es_client.helpers.utils import option_wrapper
defaults.OVERRIDE = {KEY: NEWVALUE}
click_opt_wrap = option_wrapper()

@click.group(context_settings=context_settings())
@click_opt_wrap(*cli_opts('OPTION1'))
@click_opt_wrap(*cli_opts('OPTION2', settings=defaults.OVERRIDE))
...
@click_opt_wrap(*cli_opts('OPTIONX'))
@click.pass_context
def run(ctx, OPTION1, OPTION2, ..., OPTIONX):
    # code here

The default setting KEY of OPTION2 would be overriden by NEWVALUE.

Logging¶

Logging Helpers

class es_client.helpers.logging.Whitelist(*whitelist: list)¶

Child class inheriting logging.Filter, patched to permit only specifically named loggers to write logs.

Parameters:

whitelist –

List of names defined by logging.getLogger() e.g.

filter(record)¶

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class es_client.helpers.logging.Blacklist(*whitelist: list)¶

Child class inheriting Whitelist, patched to permit all but specifically named loggers to write logs.

A monkey-patched inversion of Whitelist, i.e.

Parameters:

whitelist –

List of names defined by logging.getLogger() e.g.

filter(record)¶

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class es_client.helpers.logging.JSONFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)¶

JSON message formatting

Initialize the formatter with specified format strings.

Initialize the formatter either with the specified format string, or a default as described above. Allow for specialized date formatting with the optional datefmt argument. If datefmt is omitted, you get an ISO8601-like (or RFC 3339-like) format.

Use a style parameter of ‘%’, ‘{’ or ‘$’ to specify that you want to use one of %-formatting, str.format() ({}) formatting or string.Template formatting in your format string.

Changed in version 3.2: Added the style parameter.

WANTED_ATTRS = {'funcName': 'function', 'levelname': 'loglevel', 'lineno': 'linenum', 'message': 'message', 'name': 'name'}¶

format(record: LogRecord) → str¶

Parameters:: record – The incoming log message
Return type:: json.dumps()

es_client.helpers.logging.check_logging_config(config: Dict) → Schema¶

Parameters:: config (dict) – Logging configuration data
Returns:: SchemaCheck validated logging configuration.

Ensure that the top-level key logging is in config. Set empty default dictionary if key logging is not in config.

Pass the result to SchemaCheck for full validation.

es_client.helpers.logging.configure_logging(ctx: Context) → None¶

Parameters:: ctx – The Click command context
Return type:: None

Configure logging based on a combination of ctx.obj['draftcfg'] and ctx.params.

Values in ctx.params will override anything set in ctx.obj['draftcfg']

es_client.helpers.logging.de_dot(dot_string: str, msg: str) → Dict[str, str] | None¶

Parameters:

dot_string (str) – The dotted string
msg (str) – The message

Return type:

dict

Returns:

A nested dictionary of keys with the final value being the message

Turn message and dot_string into a nested dictionary. Used by JSONFormatter

es_client.helpers.logging.deepmerge(source: Dict, destination: Dict) → Dict¶

Parameters:

source (dict) – Source dictionary
destination (dict) – Destination dictionary

Returns:

destination

Return type:

dict

Recursively merge deeply nested dictionary structure source into destination. Used by JSONFormatter

es_client.helpers.logging.get_handler(logfile: str | None) → Handler¶

Parameters:: logfile (str) – The path of a log file
Return type:: Either FileHandler or StreamHandler
Returns:: A logging handler

This function checks first to see if a file path has been provided via logfile. If so, it will return logging.Filehandler(logfile)

If this is not provided, it will then proceed to check if it is running in a Docker container, and, if so, whether it has write permissions to /proc/1/fd/1, which is the default TTY path. If so, it will return logging.Filehandler('/proc/1/fd/1'). Writing to this path permits an app using es_client to read logs by way of:

docker logs CONTAINERNAME

If neither of the prior are true, then it will return logging.StreamHandler(stream=sys.stdout), and will write to STDOUT.

es_client.helpers.logging.get_numeric_loglevel(level: str) → int¶

Parameters:

level (str) – The log level

Return type:

int

Returns:

A numeric value mapped from level. The mapping is as follows:

Log Levels¶
Level	#	Description
NOTSET	0	When set on a logger, indicates that ancestor loggers are to be consulted to determine the effective level. If that still resolves to NOTSET, then all events are logged. When set on a handler, all events are handled.
DEBUG	10	Detailed information, typically only of interest to a developer trying to diagnose a problem.
INFO	20	Confirmation that things are working as expected.
WARNING	30	An indication that something unexpected happened, or that a problem might occur in the near future (e.g. ‘disk space low’). The software is still working as expected.
ERROR	40	Due to a more serious problem, the software has not been able to perform some function.
CRITICAL	50	A serious error, indicating that the program itself may be unable to continue running.

Raises a ValueError exception if an invalid value for level is provided.

es_client.helpers.logging.is_docker() → bool¶

Return type:: bool
Returns:: Boolean result of whether we are runinng in a Docker container or not

es_client.helpers.logging.override_logging(ctx: Context) → Dict¶

Parameters:: ctx – The Click command context
Returns:: Log configuration ready for validation

Get logging configuration from ctx.obj[‘draftcfg’] and override with any command-line options

es_client.helpers.logging.check_log_opts(log_opts: Dict) → Dict¶

Parameters:: log_opts – Logging configuration data
Returns:: Updated log_opts dictionary with default values where unset

es_client.helpers.logging.set_logging(options: Dict, logger_name: str = 'es_client') → None¶

Parameters:

options – Logging configuration data
logger_name – Default logger name to use in logging.getLogger()

Configure global logging options from options and set a default logger_name

SchemaCheck¶

SchemaCheck class and associated functions

es_client.helpers.schemacheck.password_filter(data: Dict) → Dict¶

Parameters:: data – Configuration data
Returns:: A deepcopy of data with the value obscured by REDACTED if the key is one of KEYS_TO_REDACT.

Recursively look through all nested structures of data for keys from KEYS_TO_REDACT and redact the value with REDACTED

class es_client.helpers.schemacheck.SchemaCheck(config: Dict, schema: Schema, test_what: str, location: str)¶

Parameters:

config (dict) – A configuration dictionary.
schema (Schema) – A voluptuous schema definition
test_what (str) – which configuration block is being validated
location (str) – An string to report which configuration sub-block is being tested.

Validate config with the provided Schema. test_what and location are used for reporting in case of failure. If validation is successful, the result() method returns config.

config¶: Object attribute that gets the value of param config

schema¶: Object attribute that gets the value of param schema

test_what¶: Object attribute that gets the value of param test_what

location¶: Object attribute that gets the value of param location

badvalue¶: Object attribute that is initialized with the value no bad value yet

error¶: Object attribute that is initialized with the value No error yet

parse_error() → Any¶: Report the error, and try to report the bad key or value as well.

result() → Schema¶

Return type:: Schema
Returns:: config

If validation is successful, return the value of config

If unsuccessful, try to parse the error in parse_error() and raise a FailedValidation exception.

Utils¶

Helper Utility Functions

es_client.helpers.utils.check_config(config: dict, quiet: bool = False) → dict¶

Parameters:: config – The configuration
Returns:: A validated configuration dictionary for Builder

Ensure that the top-level key elasticsearch and its sub-keys, other_settings and client are contained in config before passing it (or empty defaults) to SchemaCheck for value validation.

es_client.helpers.utils.ensure_list(data) → list¶

Parameters:: data – A list or scalar variable to act upon

Return a list, even if data is a single value

es_client.helpers.utils.file_exists(file: str) → bool¶

Parameters:: file – The file to test

Verify file exists

es_client.helpers.utils.get_version(client: Elasticsearch) → Tuple¶

Parameters:: client (Elasticsearch) – An Elasticsearch client object
Returns:: The Elasticsearch version as a 3-part tuple, (major, minor, patch)

Get the Elasticsearch version of the connected node

es_client.helpers.utils.get_yaml(path: str) → Dict¶

Parameters:: path – The path to a YAML configuration file.
Returns:: The contents of path translated from YAML to dict

Read the file identified by path and import its YAML contents.

es_client.helpers.utils.option_wrapper() → Callable¶: passthrough() the click.option() decorator function.

es_client.helpers.utils.parse_apikey_token(token: str) → Tuple¶

Parameters:: token – The base64 encoded API Key
Returns:: A tuple of (id, api_key)

Split a base64 encoded API Key Token into id and api_key

es_client.helpers.utils.passthrough(func) → Callable¶: Wrapper to make it easy to store click configuration elsewhere

es_client.helpers.utils.prune_nones(mydict: Dict) → Dict¶

Parameters:: mydict – The dictionary to act on

Remove keys from mydict whose values are None

es_client.helpers.utils.read_file(myfile: str) → str¶

Parameters:: myfile – A file to read.

Read a file and return the resulting data. Raise an ConfigurationError exception if the file is unable to be read.

es_client.helpers.utils.verify_ssl_paths(args: Dict) → None¶

Parameters:: args – The client block of the config dictionary.

Verify that the various certificate/key paths are readable. The read_file() function will raise a ConfigurationError if a file fails to be read.

es_client.helpers.utils.verify_url_schema(url: str) → str¶

Parameters:: url – The url to verify
Returns:: Verified URL

Ensure that a valid URL schema (HTTP[S]://URL:PORT) is used

Raise a ConfigurationError exception if a URL schema is invalid for any reason.

Helpers¶

Config¶

Logging¶

SchemaCheck¶

Utils¶

es_client

Navigation

Related Topics