Submodules¶

Config¶

Command-line configuration parsing and client builder helper functions

This module provides functions to manage configuration for es_client, including CLI option setup, configuration loading from YAML files, merging CLI and config settings, and creating Elasticsearch clients. It integrates with click for command-line interfaces and supports environment variables with the ENV_VAR_PREFIX prefix.

Functions:: cli_opts: Generate click option tuples for CLI decorators. cloud_id_override: Remove hosts from config if cloud_id is provided via CLI. context_settings: Configure click context with environment variable support. generate_configdict: Merge CLI and config file settings. get_arg_objects: Initialize client_args and other_args in click context. get_client: Create an Elasticsearch client using Builder. get_config: Load configuration from a YAML file or defaults. get_hosts: Validate and return a list of host URLs. get_width: Return terminal width for click context settings. hosts_override: Remove cloud_id from config if hosts is provided via CLI. options_from_dict: Decorator to add CLI options from a dictionary. override_client_args: Override client_args with CLI parameters. override_other_args: Override other_args with CLI parameters. override_settings: Merge override settings into a base settings dictionary.

es_client.config.cli_opts(value, settings=None, onoff=None, override=None)[source]¶

Generate click option tuples for CLI decorators.

Parameters:

value (str) – Option name, must be in settings or CLICK_SETTINGS.
settings (dict, optional) – Dictionary of click option parameters. Defaults to CLICK_SETTINGS.
onoff (dict, optional) – Dictionary with ‘on’ and ‘off’ keys for boolean flags (e.g., {‘on’: ‘’, ‘off’: ‘no-‘}).
override (dict, optional) – Dictionary to override settings values.

Returns:

A tuple of (option names, parameters) for click.option(), e.g.,: ((’–option’,), {‘type’: str, ‘help’: ‘Description’}).

Return type:

tuple

Raises:

ConfigurationError – If value is not in settings, settings is not a dict, or onoff parsing fails.

Example

>>> opts = cli_opts('hosts', settings={'hosts': {'type': str, 'help': 'Hosts'}})
>>> opts[0][0]
'--hosts'
>>> cli_opts('invalid', settings={'valid': {}})
Traceback (most recent call last):
    ...
es_client.exceptions.ConfigurationError: "invalid" not in settings

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.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.config.cloud_id_override(args, ctx)[source]¶

Remove hosts from config if cloud_id is provided via CLI.

Parameters:

args (dict) – Parameters from ctx.params.
ctx (click.Context) – Click command context.

Returns:

Updated args with hosts removed if cloud_id is present.

Return type:

dict

Ensures command-line cloud_id supersedes config file hosts, as they are mutually exclusive. Updates ctx.obj['client_args'].

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), obj={'client_args':
DotMap({'hosts': ['http://localhost']})})
>>> ctx.params = {'cloud_id': 'my_cloud_id'}
>>> args = {'hosts': ['http://localhost']}
>>> cloud_id_override(args, ctx)
{}
>>> ctx.obj['client_args'].hosts is None
True

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.config.context_settings()[source]¶

Configure click context settings.

Returns:

Settings for click.Command context, including terminal width,: help options, default config, and environment variable prefix.

Return type:

dict

Combines terminal width from get_width(), help options (-h, --help), default config (None), and ENV_VAR_PREFIX for environment variables.

Example

>>> settings = context_settings()
>>> settings['auto_envvar_prefix']
'ESCLIENT'
>>> 'max_content_width' in settings
True

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.config.generate_configdict(ctx)[source]¶

Merge CLI and config file settings.

Combines settings from ctx.obj['draftcfg'] and ctx.params, with CLI parameters taking precedence. Stores the result in ctx.obj['configdict'] for Builder.

Parameters:: ctx (click.Context) – Click context with draft config and parameters.
Return type:: None

Example

>>> from click import Context, Command
>>> cfg = {'client': {'hosts': ['http://localhost']}, 'other_settings': {}}
>>> ctx = Context(Command('cmd'), obj={'draftcfg': {'elasticsearch': cfg}})
>>> ctx.params = {'hosts': ['http://127.0.0.1:9200'], 'config': None}
>>> generate_configdict(ctx)
>>> ctx.obj['configdict']['elasticsearch']['client']['hosts']
['http://127.0.0.1:9200']

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.config.get_arg_objects(ctx)[source]¶

Initialize client_args and other_args in click context.

Sets ctx.obj['client_args'] and ctx.obj['other_args'] as dotmap.DotMap objects, populating them from ctx.obj['draftcfg'] via check_config().

Parameters:: ctx (click.Context) – Click context with draft configuration.
Return type:: None

Example

>>> from click import Context, Command
>>> cfg = {'client': {'hosts': ['http://localhost']}, 'other_settings': {}}
>>> ctx = Context(Command('cmd'), obj={'draftcfg': {'elasticsearch': cfg}})
>>> get_arg_objects(ctx)
>>> ctx.obj['client_args'].hosts
['http://localhost']

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.config.get_client(configdict=None, configfile=None, autoconnect=False, version_min=(9, 0, 0), version_max=(9, 99, 99))[source]¶

Create an Elasticsearch client using Builder.

Parameters:

configdict (dict, optional) – Configuration dictionary with ‘elasticsearch’ key.
configfile (str, optional) – Path to a YAML configuration file.
autoconnect (bool, optional) – Connect to client automatically. Defaults to False.
version_min (tuple, optional) – Minimum Elasticsearch version. Defaults to VERSION_MIN.
version_max (tuple, optional) – Maximum Elasticsearch version. Defaults to VERSION_MAX.

Returns:

Configured Elasticsearch client.

Return type:

elasticsearch9.Elasticsearch

Raises:

ESClientException – If client connection fails.
ConfigurationError – If configuration is invalid.

Prioritizes configdict over configfile. Uses defaults if neither is provided.

Example

>>> config = {'elasticsearch': {'client': {'hosts': ['http://localhost:9200']}}}
>>> client = get_client(configdict=config)
>>> isinstance(client, Elasticsearch)
True

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.config.get_config(ctx, quiet=True)[source]¶

Load configuration from a YAML file or defaults.

Checks ctx.params['config'] for a YAML file path, falling back to ctx.obj['default_config']. Stores the result in ctx.obj['draftcfg'].

Parameters:

ctx (click.Context) – Click context to store configuration.
quiet (bool, optional) – Suppress stdout messages for default config. Defaults to True.

Returns:

Updated context with ‘draftcfg’ set.

Return type:

click.Context

Raises:

OSError – If the YAML file cannot be read.
yaml.YAMLError – If the YAML file is invalid.

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), obj={'default_config': None},
params={'config': None})
>>> ctx = get_config(ctx, quiet=True)
>>> 'draftcfg' in ctx.obj
True

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.config.get_hosts(ctx)[source]¶

Validate and return a list of host URLs.

Retrieves hosts from ctx.params['hosts'] and validates their URL schemas using verify_url_schema().

Parameters:: ctx (click.Context) – Click context with host parameters.
Returns:: List of validated host URLs, or None if no hosts are provided.
Return type:: list or None
Raises:: ConfigurationError – If URL schema validation fails.

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), params={'hosts': ['http://localhost:9200']})
>>> hosts = get_hosts(ctx)
>>> hosts
['http://localhost:9200']

es_client.config.get_width()[source]¶

Return terminal width for click context settings.

Returns:

Dictionary with ‘max_content_width’ set to terminal width from: shutil.get_terminal_size().

Return type:

dict

Example

>>> width = get_width()
>>> 'max_content_width' in width
True

es_client.config.hosts_override(args, ctx)[source]¶

Remove cloud_id from config if hosts is provided via CLI.

Parameters:

args (dict) – Parameters from ctx.params.
ctx (click.Context) – Click command context.

Returns:

Updated args with cloud_id removed if hosts is present.

Return type:

dict

Ensures command-line hosts supersedes config file cloud_id, as they are mutually exclusive. Updates ctx.obj['client_args'].

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), obj={'client_args': DotMap({'cloud_id': 'my_cloud_id'})})
>>> ctx.params = {'hosts': ['http://localhost']}
>>> args = {'cloud_id': 'my_cloud_id'}
>>> hosts_override(args, ctx)
{}
>>> ctx.obj['client_args'].cloud_id is None
True

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.

es_client.config.options_from_dict(options_dict)[source]¶

Decorator to add CLI options from a dictionary.

Parameters:: options_dict (dict) – Dictionary of option names and their click parameters.
Returns:: Decorator function to apply click options to a command.
Return type:: callable

Example

>>> opts = {'hosts': {'settings': {'type': str, 'help': 'Hosts'}}}
>>> @options_from_dict(opts)
... def cmd(hosts): pass
>>> hasattr(cmd, '__click_params__')
True

es_client.config.override_client_args(ctx)[source]¶

Override client_args with CLI parameters.

Updates ctx.obj['client_args'] with values from ctx.params for keys in config_settings(). Sets default hosts if neither hosts nor cloud_id is provided.

Parameters:: ctx (click.Context) – Click context with parameters and client_args.
Return type:: None

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), obj={'client_args': DotMap()})
>>> ctx.params = {'hosts': ['http://127.0.0.1:9200'], 'config': None}
>>> override_client_args(ctx)
>>> ctx.obj['client_args'].hosts
['http://127.0.0.1:9200']

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.config.override_other_args(ctx)[source]¶

Override other_args with CLI parameters.

Updates ctx.obj['other_args'] with values from ctx.params for non-client settings (e.g., master_only, username, api_key).

Parameters:: ctx (click.Context) – Click context with parameters and other_args.
Return type:: None

Example

>>> from click import Context, Command
>>> ctx = Context(Command('cmd'), obj={'other_args': DotMap()})
>>> ctx.params = {
    'username': 'user',
    'password': 'pass',
    'id': None,
    'api_key': None,
    'api_token': None,
    'config': None
}
>>> override_other_args(ctx)
>>> ctx.obj['other_args'].username
'user'

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

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

es_client.config.override_settings(settings, override)[source]¶

Merge override settings into a base settings dictionary.

Parameters:

settings (dict) – Base settings dictionary.
override (dict) – Settings to override settings.

Returns:

Updated settings with override values.

Return type:

dict

Raises:

ConfigurationError – If override is not a dict.

Used by cli_opts() and options_from_dict() to customize click options.

Example

>>> settings = {'type': str, 'help': 'Hosts'}
>>> override = {'help': 'Updated Hosts'}
>>> override_settings(settings, override)
{'type': <class 'str'>, 'help': 'Updated Hosts'}

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

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.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 configuration for es_client

This module provides utilities for configuring logging in es_client, including filters (Whitelist, Blacklist) and a JSON formatter (JSONFormatter). It supports log level conversion, logger name normalization, and setup for CLI and configuration file inputs, integrating with click.Context.

Classes:: Whitelist: Logging filter to allow specific logger names. Blacklist: Logging filter to block specific logger names. JSONFormatter: Custom formatter for JSON log output.
Functions:: check_logging_config: Validate logging configuration using SchemaCheck. configure_logging: Configure logging from a click context. de_dot: Replace dots in logger names with underscores. deepmerge: Recursively merge dictionaries for JSON formatting. get_format_string: Return a log format string based on log level. get_logger: Set up the root logger with handlers and filters. get_numeric_loglevel: Convert a string log level to a logging constant. override_logging: Merge CLI and config file logging settings. check_log_opts: Apply default logging options. set_logging: Configure global logging with handlers and filters.

es_client.logging.check_logging_config(config)[source]¶

Validate logging configuration using SchemaCheck.

Parameters:

config (dict) – Logging configuration data.

Returns:

Validated logging configuration from: SchemaCheck.

Return type:

voluptuous.Schema

Ensures the top-level key logging is in config. Sets an empty default dictionary if logging is absent. Passes the result to SchemaCheck for validation against config_logging().

Raises:: TypeError – If config is not a dictionary.
Parameters:: config (Dict)
Return type:: Schema

Example

>>> config = {'logging': {'loglevel': 'INFO'}}
>>> result = check_logging_config(config)
>>> result['loglevel']
'INFO'
>>> config = {}
>>> result = check_logging_config(config)
>>> result
{}

es_client.logging.configure_logging(ctx)[source]¶

Configure logging based on a click context.

Merges logging settings from ctx.obj['draftcfg'] and ctx.params, with CLI parameters taking precedence. Validates and applies the merged settings using set_logging().

Parameters:: ctx (click.Context) – Click command context containing logging config.
Return type:: None

Example

>>> from click import Context, Command
>>> cfg = {
    'logging': {'loglevel': 'INFO', 'logfile': None, 'logformat': 'default'}
}
>>> ctx = Context(Command('cmd'), obj={'draftcfg': cfg}, params={})
>>> configure_logging(ctx)
>>> logging.getLogger('').level
20

es_client.logging.de_dot(dot_string, msg)[source]¶

Convert a dotted string and message into a nested dictionary.

Parameters:

dot_string (str) – Dotted string (e.g., ‘es_client.utils’).
msg (str) – Message to nest under the final key.

Returns:

Nested dictionary with msg as the leaf value.

Return type:

dict

Raises:

LoggingException – If dictionary creation fails.

Used by JSONFormatter to structure log data.

Example

>>> de_dot('es_client.utils', 'test')
{'es_client': {'utils': 'test'}}
>>> de_dot('simple', 'test')
{'simple': 'test'}

es_client.logging.deepmerge(source, destination)[source]¶

Recursively merge a source dictionary into a destination dictionary.

Parameters:

source (dict) – Source dictionary to merge.
destination (dict) – Destination dictionary to update.

Returns:

Updated destination dictionary.

Return type:

dict

Used by JSONFormatter to combine log attributes.

Example

>>> source = {'a': {'b': 1}, 'c': 2}
>>> destination = {'a': {'d': 3}, 'e': 4}
>>> deepmerge(source, destination)
{'a': {'b': 1, 'd': 3}, 'e': 4, 'c': 2}

es_client.logging.get_format_string(nll)[source]¶

Return a log format string based on the numeric log level.

Parameters:: nll (int) – Numeric log level (e.g., 10 for DEBUG, 20 for INFO).
Returns:: Format string for logging.Formatter.
Return type:: str

Example

>>> get_format_string(10)  # DEBUG
'%(asctime)s %(levelname)-9s %(name)30s %(funcName)23s:%(lineno)-4d %(message)s'
>>> get_format_string(20)  # INFO
'%(asctime)s %(levelname)-9s %(message)s'

es_client.logging.get_logger(log_opts)[source]¶

Configure the root logger with appropriate handlers.

If logfile is provided in log_opts, uses a logging.FileHandler. Otherwise, splits logs into logging.StreamHandler instances for stdout (up to INFO) and stderr (WARNING and above). Applies formatters and filters based on logformat and blacklist.

Parameters:

log_opts (dict) – Logging configuration with keys: loglevel, logfile, logformat, blacklist.

Raises:

OSError – If logfile cannot be opened.
ValueError – If loglevel is invalid.

Return type:

None

Example

>>> log_opts = {
    'loglevel': 'INFO',
    'logfile': None,
    'logformat': 'default',
    'blacklist': []
}
>>> get_logger(log_opts)
>>> len(logging.getLogger('').handlers) >= 1
True

es_client.logging.get_numeric_loglevel(level)[source]¶

Convert a string log level to a logging module constant.

Parameters:: level (str) – Log level name (e.g., ‘DEBUG’, ‘INFO’).
Returns:: Corresponding logging constant (e.g., logging.DEBUG).
Return type:: int
Raises:: ValueError – If the level is not a valid log level.

The mapping is:

Log Levels¶
Level	#	Description
NOTSET	0	When set on a logger, ancestor loggers determine the effective level. If NOTSET, all events are logged. On a handler, all events are handled.
DEBUG	10	Detailed information for diagnosing problems.
INFO	20	Confirmation that things are working as expected.
WARNING	30	Indicates an unexpected issue or potential problem (e.g., ‘disk space low’).
ERROR	40	A serious problem preventing some functionality.
CRITICAL	50	A severe error, possibly halting the program.

Example

>>> get_numeric_loglevel('DEBUG')
10
>>> get_numeric_loglevel('INFO')
20
>>> get_numeric_loglevel('INVALID')
Traceback (most recent call last):
    ...
ValueError: Invalid log level: INVALID

es_client.logging.override_logging(ctx)[source]¶

Merge CLI and config file logging settings.

Retrieves logging configuration from ctx.obj['draftcfg'] and overrides with ctx.params, with CLI taking precedence. Validates the config using check_logging_config().

Parameters:: ctx (click.Context) – Click command context.
Returns:: Merged and validated logging configuration.
Return type:: dict

Example

>>> from click import Context, Command
>>> cfg = {'logging': {'loglevel': 'INFO'}}
>>> ctx = Context(
    Command('cmd'), obj={'draftcfg': cfg}, params={'loglevel': 'DEBUG'}
)
>>> result = override_logging(ctx)
>>> result['loglevel']
'DEBUG'

es_client.logging.check_log_opts(log_opts)[source]¶

Apply default logging options to unset keys.

Parameters:

log_opts (dict) – Logging configuration data.

Returns:

Updated log_opts with defaults from: LOGDEFAULTS.

Return type:

dict

Example

>>> log_opts = {'loglevel': 'INFO'}
>>> result = check_log_opts(log_opts)
>>> result['loglevel']
'INFO'
>>> 'logfile' in result
True

es_client.logging.set_logging(options)[source]¶

Configure global logging options.

Applies settings from options to the root logger, attaching handlers and filters. Adds a logging.NullHandler for the ‘elasticsearch9.trace’ logger to suppress trace logs.

Parameters:

options (dict) – Logging configuration with keys: loglevel, logfile, logformat, blacklist.

Raises:

OSError – If logfile cannot be opened.
ValueError – If loglevel is invalid.

Return type:

None

Example

>>> options = {
    'loglevel': 'INFO',
    'logfile': None,
    'logformat': 'default',
    'blacklist': []
}
>>> set_logging(options)
>>> len(logging.getLogger('').handlers) >= 1
True

class es_client.logging.Whitelist[source]¶

Bases: Filter

Logging filter to allow only specified logger names.

Parameters:: *whitelist (list) – Logger names to allow (e.g., [‘es_client’]).

whitelist¶

List of logging.Filter objects for allowed names.

Type:: list

Example

>>> import logging
>>> record = logging.makeLogRecord({'name': 'es_client.test', 'msg': 'Test'})
>>> whitelist = Whitelist(['es_client'])
>>> whitelist.filter(record)
True
>>> record = logging.makeLogRecord({'name': 'other.test', 'msg': 'Test'})
>>> whitelist.filter(record)
False

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

__init__(*whitelist)[source]¶

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

Parameters:: whitelist (List[str])

filter(record)[source]¶

Filter log records based on logger name.

Parameters:: record (LogRecord)
Return type:: bool

class es_client.logging.Blacklist[source]¶

Bases: Whitelist

Logging filter to block specified logger names.

Inherits from Whitelist, inverting the filter logic to block listed names.

Parameters:: *blacklist (list) – Logger names to block (e.g., [‘es_client.utils’]).

whitelist¶

List of logging.Filter objects for blocked names.

Type:: list

Example

>>> import logging
>>> record = logging.makeLogRecord({'name': 'es_client.test', 'msg': 'Test'})
>>> blacklist = Blacklist(['es_client.test'])
>>> blacklist.filter(record)
False
>>> record = logging.makeLogRecord({'name': 'other.test', 'msg': 'Test'})
>>> blacklist.filter(record)
True

Initialize a filter.

Initialize with the name of the logger which, together with its children, will have its events allowed through the filter. If no name is specified, allow every event.

filter(record)[source]¶

Filter log records based on logger name.

Parameters:: record (LogRecord)
Return type:: bool

class es_client.logging.JSONFormatter[source]¶

Bases: Formatter

Custom logging formatter for JSON output.

Formats log records as JSON objects with timestamp, log level, logger name, function, line number, and message.

Example

>>> import logging
>>> record = logging.makeLogRecord({
...     'name': 'test', 'levelname': 'INFO', 'msg': 'Test message',
...     'funcName': 'test_func', 'lineno': 42, 'created': 1625097600.0,
...     'msecs': 123.456
... })
>>> formatter = JSONFormatter()
>>> formatted = formatter.format(record)
>>> 'test' in formatted and 'INFO' in formatted
True

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)[source]¶

Format a log record as a JSON string.

Parameters:: record (LogRecord)
Return type:: str

SchemaCheck¶

Schema validation and redaction utilities

This module provides the SchemaCheck class to validate configuration dictionaries against a voluptuous.Schema and a function to redact sensitive data for secure logging. It supports configuration validation for Builder and logging in logging.

Classes:: SchemaCheck: Validates a configuration dictionary against a schema.
Functions:: password_filter: Redact sensitive values from a configuration dictionary.

es_client.schemacheck.password_filter(data)[source]¶

Redact sensitive values from a configuration dictionary.

Parameters:

data (dict) – Configuration dictionary to process.

Returns:

A deep copy of data with sensitive values (keys in: KEYS_TO_REDACT) replaced with ‘REDACTED’.

Return type:

dict

Recursively traverses data, replacing values of keys listed in KEYS_TO_REDACT (e.g., ‘password’, ‘api_key’) with ‘REDACTED’ for secure logging.

Example

>>> data = {'user': 'test', 'password': 'secret', 'nested': {'api_key': 'key'}}
>>> filtered = password_filter(data)
>>> filtered
{'user': 'test', 'password': 'REDACTED', 'nested': {'api_key': 'REDACTED'}}
>>> data['password']  # Original unchanged
'secret'

class es_client.schemacheck.SchemaCheck[source]¶

Bases: object

Validate a configuration dictionary against a voluptuous schema.

Parameters:

config (dict) – Configuration dictionary to validate.
schema (voluptuous.Schema) – Schema to validate against.
test_what (str) – Description of the configuration block (e.g., ‘Client Config’).
location (str) – Context of the configuration (e.g., ‘elasticsearch.client’).

config¶

The configuration dictionary.

Type:: dict

schema¶

The validation schema.

Type:: voluptuous.Schema

test_what¶

Description of the configuration block.

Type:: str

location¶

Context of the configuration.

Type:: str

badvalue¶

Invalid value causing validation failure, or ‘no bad value yet’.

Type:: str

error¶

Validation error message, or ‘No error yet’.

Type:: str

Raises:: FailedValidation – If validation fails.

Example

>>> from voluptuous import Schema
>>> schema = Schema({'host': str})
>>> config = {'host': 'localhost'}
>>> check = SchemaCheck(config, schema, 'Test Config', 'test')
>>> check.result()
{'host': 'localhost'}
>>> config = {'host': 123}
>>> check = SchemaCheck(config, schema, 'Test Config', 'test')
>>> check.result()
Traceback (most recent call last):
    ...
es_client.exceptions.FailedValidation: Configuration: Test Config: Location:
test: Bad Value: "123", expected str @ data['host']. Check configuration file.

__init__(config, schema, test_what, location)[source]¶

Parameters:

config (Dict)
schema (Schema)
test_what (str)
location (str)

parse_error()[source]¶

Extract and report the invalid value causing a validation error.

Attempts to parse the error message to identify the bad value, updating badvalue. Logs errors if parsing fails.

Returns:: Updates badvalue and logs the result.
Return type:: None

Example

>>> from voluptuous import Schema
>>> schema = Schema({'host': str})
>>> config = {'host': 123}
>>> check = SchemaCheck(config, schema, 'Test Config', 'test')
>>> try:
...     check.result()
... except FailedValidation:
...     check.badvalue
'123'

result()[source]¶

Validate the configuration and return the result.

Returns:

Validated configuration from: config if successful.

Return type:

voluptuous.Schema

Raises:

FailedValidation – If validation fails, including error details and bad value.

Calls parse_error() to extract the invalid value if validation fails.

Example

>>> from voluptuous import Schema
>>> schema = Schema({'host': str})
>>> config = {'host': 'localhost'}
>>> check = SchemaCheck(config, schema, 'Test Config', 'test')
>>> check.result()
{'host': 'localhost'}

Utils¶

Utility functions for es_client

This module provides helper functions for configuration validation, file handling, and CLI setup in es_client. It supports Builder, logging, and config with tasks like URL validation, YAML parsing, and sensitive data redaction.

Functions:: check_config: Validate an Elasticsearch configuration dictionary. ensure_list: Convert a scalar or list to a list. file_exists: Check if a file exists. get_version: Retrieve the Elasticsearch version as a tuple. get_yaml: Load a YAML file with environment variable parsing. option_wrapper: Wrap click.option for configuration storage. parse_apikey_token: Decode a base64-encoded API key token. passthrough: Generic wrapper for click decorators. prune_nones: Remove keys with None values from a dictionary. read_file: Read a file’s contents. verify_ssl_paths: Verify readability of SSL certificate/key paths. verify_url_schema: Validate and normalize URL schemas for hosts.

es_client.utils.check_config(config, quiet=False)[source]¶

Validate an Elasticsearch configuration dictionary.

Parameters:

config (dict) – Configuration dictionary to validate.
quiet (bool, optional) – Suppress warning logs if True. Defaults to False.

Returns:

Validated configuration for Builder.

Return type:

dict

Ensures ‘elasticsearch’, ‘client’, and ‘other_settings’ keys are present in config, using ES_DEFAULT if missing. Validates against config_schema() via SchemaCheck.

Raises:

FailedValidation – If validation fails.

Parameters:

config (dict)
quiet (bool)

Return type:

dict

Example

>>> config = {'elasticsearch': {'client': {'hosts': ['http://localhost:9200']}}}
>>> result = check_config(config, quiet=True)
>>> result['client']['hosts']
['http://localhost:9200']
>>> check_config([], quiet=True)
{'client': {}, 'other_settings': {}}

es_client.utils.ensure_list(data)[source]¶

Convert a scalar or list to a list.

Parameters:: data (Any) – A scalar or list to convert.
Returns:: A list containing data if scalar, or data itself if already a list.
Return type:: list

Example

>>> ensure_list('item')
['item']
>>> ensure_list(['item1', 'item2'])
['item1', 'item2']

es_client.utils.file_exists(file)[source]¶

Check if a file exists.

Parameters:: file (str) – Path to the file.
Returns:: True if file exists, False otherwise.
Return type:: bool

Example

>>> from pathlib import Path
>>> Path('test.txt').write_text('test')
4
>>> file_exists('test.txt')
True
>>> file_exists('nonexistent.txt')
False
>>> Path('test.txt').unlink()

es_client.utils.get_version(client)[source]¶

Retrieve the Elasticsearch version as a tuple.

Parameters:: client (elasticsearch9.Elasticsearch) – Elasticsearch client instance.
Returns:: Version as (major, minor, patch).
Return type:: tuple

Extracts the version from client.info()[‘version’][‘number’], stripping non-semantic tags (e.g., ‘-dev’, ‘-beta’) and limiting to three parts.

Example

>>> from unittest.mock import Mock
>>> client = Mock()
>>> client.info.return_value = {'version': {'number': '9.0.0'}}
>>> get_version(client)
(9, 0, 0)

es_client.utils.get_yaml(path)[source]¶

Load a YAML file with environment variable parsing.

Parameters:: path (str) – Path to the YAML file.
Returns:: Parsed YAML contents.
Return type:: dict
Raises:: ConfigurationError – If the YAML file is invalid or unreadable.

Supports environment variable substitution in YAML using ${VAR:DEFAULT} syntax.

Example

>>> from pathlib import Path
>>> Path('test.yaml').write_text('host: ${HOST:localhost}')
23
>>> os.environ['HOST'] = '127.0.0.1'
>>> config = get_yaml('test.yaml')
>>> config['host']
'127.0.0.1'
>>> Path('test.yaml').unlink()

es_client.utils.option_wrapper()[source]¶

Wrap click.option for configuration storage.

Returns:: Wrapper function for click.option().
Return type:: callable

Example

>>> wrapper = option_wrapper()
>>> isinstance(wrapper(('option',), {}), click.decorators._Option)
True

es_client.utils.parse_apikey_token(token)[source]¶

Decode a base64-encoded API key token.

Parameters:: token (str) – Base64-encoded token in the format ‘id:api_key’.
Returns:: (id, api_key) extracted from the token.
Return type:: tuple
Raises:: ConfigurationError – If the token is invalid.

Example

>>> parse_apikey_token('Zm9vOmJhcg==')  # base64 for 'foo:bar'
('foo', 'bar')
>>> parse_apikey_token('invalid')
Traceback (most recent call last):
    ...
es_client.exceptions.ConfigurationError: Unable to parse base64 API Key
Token: Incorrect padding

es_client.utils.passthrough(func)[source]¶

Generic wrapper for click decorators.

Parameters:: func (callable) – Function to wrap (e.g., click.option()).
Returns:: Wrapped function passing arguments through.
Return type:: callable

Example

>>> wrapped = passthrough(lambda x: x * 2)
>>> wrapped(5)
10

es_client.utils.prune_nones(mydict)[source]¶

Remove keys with None values from a dictionary.

Parameters:: mydict (dict) – Dictionary to process.
Returns:: Dictionary with None-valued keys removed.
Return type:: dict

Example

>>> prune_nones({'a': 1, 'b': None, 'c': 'None'})
{'a': 1}

es_client.utils.read_file(myfile)[source]¶

Read a file’s contents.

Parameters:: myfile (str) – Path to the file.
Returns:: File contents.
Return type:: str
Raises:: ConfigurationError – If the file cannot be read.

Example

>>> from pathlib import Path
>>> Path('test.txt').write_text('test')
4
>>> read_file('test.txt')
'test'
>>> Path('test.txt').unlink()

es_client.utils.verify_ssl_paths(args)[source]¶

Verify readability of SSL certificate/key paths.

Parameters:: args (dict) – Configuration with ‘ca_certs’, ‘client_cert’, or ‘client_key’ keys.
Raises:: ConfigurationError – If any file is unreadable.
Return type:: None

Checks if paths in args for ‘ca_certs’, ‘client_cert’, or ‘client_key’ are readable using read_file().

Example

>>> from pathlib import Path
>>> Path('cert.pem').write_text('cert')
4
>>> verify_ssl_paths({'ca_certs': 'cert.pem'})
>>> verify_ssl_paths({'ca_certs': 'nonexistent.pem'})
Traceback (most recent call last):
    ...
es_client.exceptions.ConfigurationError: Unable to read file nonexistent.pem.
Exception: [Errno 2] No such file or directory: 'nonexistent.pem'
>>> Path('cert.pem').unlink()

es_client.utils.verify_url_schema(url)[source]¶

Validate and normalize a URL schema for Elasticsearch hosts.

Parameters:: url (str) – URL to validate (e.g., ‘http://localhost:9200’).
Returns:: Normalized URL with schema and port (e.g., ‘http://localhost:80’).
Return type:: str
Raises:: ConfigurationError – If the URL schema is invalid.

Ensures the URL uses ‘http’ or ‘https’ and includes a port (defaults to 80 for http, 443 for https if omitted).

Example

>>> verify_url_schema('https://localhost')
'https://localhost:443'
>>> verify_url_schema('ftp://localhost')
Traceback (most recent call last):
    ...
es_client.exceptions.ConfigurationError: URL Schema invalid for ftp://localhost