API Reference

Automatic documentation for the public API exported by the package.

ES Query Generator - Build complex Elasticsearch queries from simple Python dictionaries.

class es_query_gen.ESClientSingleton

Bases: object

Lightweight singleton/registry for Elasticsearch clients (sync and async).

Use connect() to create and register a default sync client. Use connect_async() to create and register a default async client. Use set(), get(), clear() for sync client management. Use set_async(), get_async(), clear_async() for async client management.

classmethod clear(client_key: str | None = None) → None

Clear the default synchronous Elasticsearch client. Args:

client_key: Key to identify the synchronous client (default: None, clears all).

classmethod clear_async(client_key: str | None = None) → None

Clear the default asynchronous Elasticsearch client. Args:

client_key: Key to identify the asynchronous client (default: None, clears all).

classmethod close(client_key: str | None = None) → None

Close Elasticsearch client connection(s).

If client_key is provided, closes that specific client. Otherwise, closes all registered synchronous clients.

Args:

client_key: Key to identify the synchronous client (default: None, closes all).

async classmethod close_async(client_key: str | None = None) → None

Close asynchronous Elasticsearch client connection(s).

If client_key is provided, closes that specific async client. Otherwise, closes all registered asynchronous clients.

Args:

client_key: Key to identify the asynchronous client (default: None, closes all).

classmethod connect(connection_string: str | None = None, host: str = 'localhost', port: int = 9200, scheme: str = 'http', username: str | None = None, password: str | None = None, verify_certs: bool = True, client_key: str = 'default', **kwargs: Any) → Elasticsearch

Create and register a synchronous Elasticsearch client as default.

Args:

connection_string: Full connection string (takes precedence over other params). host: Elasticsearch host (default: ‘localhost’). port: Elasticsearch port (default: 9200). scheme: Connection scheme, ‘http’ or ‘https’ (default: ‘http’). username: Username for authentication (optional). password: Password for authentication (optional). verify_certs: Whether to verify SSL certificates (default: True). client_key: Key to identify the synchronous client (default: ‘default’). kwargs: Additional arguments to pass to Elasticsearch client.

Returns:

The created Elasticsearch client instance.

classmethod connect_async(connection_string: str | None = None, host: str = 'localhost', port: int = 9200, scheme: str = 'http', username: str | None = None, password: str | None = None, verify_certs: bool = True, client_key: str = 'default', **kwargs: Any) → AsyncElasticsearch

Create and register an asynchronous Elasticsearch client as default.

Args:

connection_string: Full connection string (takes precedence over other params). host: Elasticsearch host (default: ‘localhost’). port: Elasticsearch port (default: 9200). scheme: Connection scheme, ‘http’ or ‘https’ (default: ‘http’). username: Username for authentication (optional). password: Password for authentication (optional). verify_certs: Whether to verify SSL certificates (default: True). client_key: Key to identify the asynchronous client (default: ‘default’). kwargs: Additional arguments to pass to AsyncElasticsearch client.

Returns:

The created AsyncElasticsearch client instance.

classmethod get(client_key: str = 'default') → Elasticsearch | None

Get the default synchronous Elasticsearch client.

Args:

client_key: Key to identify the synchronous client (default: ‘default’).

Returns:

The registered Elasticsearch client or None if not set.

classmethod get_async(client_key: str = 'default') → AsyncElasticsearch | None

Get the default asynchronous Elasticsearch client.

Args:

client_key: Key to identify the asynchronous client (default: ‘default’).

Returns:

The registered AsyncElasticsearch client or None if not set.

classmethod set(client: Elasticsearch, client_key: str = 'default') → None

Set the default synchronous Elasticsearch client.

Args:

client: The Elasticsearch client instance to set as default. client_key: Key to identify the synchronous client (default: ‘default’).

classmethod set_async(client: AsyncElasticsearch, client_key: str = 'default') → None

Set the default asynchronous Elasticsearch client.

Args:

client: The AsyncElasticsearch client instance to set as default. client_key: Key to identify the asynchronous client (default: ‘default’).

class es_query_gen.ESResponseParser(query_config: QueryConfig | Dict[str, Any])

Bases: object

Parse Elasticsearch responses according to a QueryConfig.

Usage:

parser = ESResponseParser(config)
results = parser.parse_data(es_response)

Two parsing strategies are chosen automatically based on QueryConfig.aggs:

  • No aggs — extracts each document from hits.hits, merging _source fields with _id.

  • With aggs — walks the nested aggregation tree and extracts documents from the top_hits_bucket at the leaf level.

A new ESResponseParser instance should be created per request, or parse_data can be called multiple times on the same instance — results are reset at the start of every call.

parse_aggregations(response: Dict[str, Any]) → None

Walk the nested aggregation tree and append leaf documents to self.results.

Builds a linked list from self.query_config.aggs, then recursively traverses bucket levels, extracting documents from the top_hits_bucket at the deepest aggregation level.

Args:

response: The raw Elasticsearch response containing aggregations.

parse_data(response: Dict[str, Any]) → List[Dict[str, Any]]

Parse an ES response and return a list of documents.

Resets the internal results list on every call, so the same ESResponseParser instance can be safely reused across requests.

Args:

response: The raw JSON response from Elasticsearch (as a dict).

Returns:

List of dicts representing documents or aggregation results.

parse_search_results(response: Dict[str, Any]) → None

Extract hits from an ES response and append them to self.results.

Each document includes its _source fields merged with _id.

Args:

response: The raw Elasticsearch response containing hits.

class es_query_gen.FieldTransformRule(*, source_key: str, target_key: str | None = None, default: Any = None, type_cast: Literal['STR', 'INT', 'FLOAT', 'BOOL', 'NUMBER', 'EPOCH', 'DATE'] | None = None, value_casing: Literal['LOWER', 'UPPER', 'TITLE'] | None = None, date_format: str | None = None, target_date_format: str | None = None)

Bases: BaseModel

Transformation rule for a single field in a parsed document.

Attributes:

source_key: The key name as it appears in the parser output. target_key: Completely override the output key name. When provided, the

name is used verbatim — key_casing is NOT applied to it, because you have already chosen the exact name you want. Defaults to source_key when omitted (casing will then apply).

default: Value to use when the field is None or absent. type_cast: Optional coercion applied to the (possibly-defaulted) value.

Supported values:

  • "STR" / "INT" / "FLOAT" / "BOOL" — standard Python casts.

  • "NUMBER" — coerces to float.

  • "EPOCH" — converts a datetime object or ISO-8601 string to a UTC Unix timestamp (int).

key_casing: Per-field key-name casing override. Takes priority over

TransformConfig.key_casing. Ignored when target_key is explicitly set. Leave as None to inherit the global setting.

value_casing: Apply string casing to the value of this field

(only effective when the final value is a str). One of "LOWER", "UPPER", "TITLE".

Example:

# Explicit rename + value uppercased
FieldTransformRule(
    source_key="status",
    target_key="Status",       # used as-is, verbatim
    value_casing="UPPER",      # "active" → "ACTIVE"
)

# No rename — source_key used as output key
FieldTransformRule(
    source_key="created_at",
    type_cast="EPOCH",
)
date_format: str | None

strptime format for parsing the input value when type_cast='DATE'. Defaults to '%Y-%m-%dT%H:%M:%S' when not supplied.

default: Any
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

source_key: str
target_date_format: str | None

strftime format for the output string when type_cast='DATE'. - datetime input → formatted with this pattern. - str input → re-formatted from date_format to this pattern. Defaults to date_format (no-op reformat) when not supplied.

target_key: str | None
type_cast: Literal['STR', 'INT', 'FLOAT', 'BOOL', 'NUMBER', 'EPOCH', 'DATE'] | None
value_casing: Literal['LOWER', 'UPPER', 'TITLE'] | None

Apply string casing to the value of this field (only when the value is a str).

class es_query_gen.QueryBuilder

Bases: object

Build Elasticsearch queries from QueryConfig objects.

All methods are static — the class holds no instance state and does not need to be instantiated. Both calling styles work:

# Preferred (no instantiation needed) query = QueryBuilder.build(config)

# Also fine (throwaway instance, same result) query = QueryBuilder().build(config)

static build(es_query_config: QueryConfig | Dict[str, Any]) → Dict[str, Any]

Build a complete Elasticsearch query from a QueryConfig object.

Creates a fresh local query dict on every call — no shared state, fully thread-safe, and safely callable as QueryBuilder.build(config) without instantiation.

Args:
es_query_config: QueryConfig instance or a plain dict that will be

coerced via model_validate.

Returns:

Dictionary representing a complete Elasticsearch query DSL.

class es_query_gen.ResponseTransformer(config: TransformConfig)

Bases: object

Apply a TransformConfig to a list of parsed documents.

All config is validated and pre-processed in __init__ so the hot path (transform) is pure Python dict operations with no Pydantic overhead.

Usage:

config = TransformConfig(
    rules=[
        FieldTransformRule(source_key="user_id", target_key="userId"),
        FieldTransformRule(source_key="created_at", type_cast="EPOCH"),
        FieldTransformRule(source_key="score", default=0.0, type_cast="FLOAT"),
    ],
    key_casing="LOWER",
)
transformer = ResponseTransformer(config)
output = transformer.transform(parser.parse_data(response))
transform(docs: List[Dict[str, Any]]) → List[Dict[str, Any]]

Apply the transform config to every document in docs.

Args:

docs: List of dicts as returned by ESResponseParser.parse_data.

Returns:

New list of transformed dicts — the originals are not mutated.

class es_query_gen.TransformConfig(*, rules: List[FieldTransformRule] = [], include_unmapped: bool = True)

Bases: BaseModel

Configuration for ResponseTransformer.

Attributes:
rules: Per-field transformation rules (rename, default, type cast, value casing).

Fields not listed in rules are passed through unchanged when include_unmapped is True.

include_unmapped: When True (default), keys not covered by any rule

are copied to the output as-is. Set to False to strip unmapped fields from the output.

Example:

TransformConfig(
    rules=[
        FieldTransformRule(source_key="user_id", target_key="userId"),
        FieldTransformRule(source_key="score", type_cast="FLOAT", default=0.0),
        FieldTransformRule(source_key="status", value_casing="UPPER"),
    ],
)
include_unmapped: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

rules: List[FieldTransformRule]
es_query_gen.clear_es_client(client_key: str = 'default') → None

Clear the module-level default synchronous Elasticsearch client. Args:

client_key: Key to identify the synchronous client (default: ‘default’).

es_query_gen.clear_es_client_async(client_key: str = 'default') → None

Clear the module-level default asynchronous Elasticsearch client.

Args:

client_key: Key to identify the asynchronous client (default: ‘default’).

es_query_gen.close_all_es_clients() → None

Close all synchronous Elasticsearch client connections.

async es_query_gen.close_all_es_clients_async() → None

Close all asynchronous Elasticsearch client connections.

es_query_gen.close_es_client(client_key: str | None = 'default') → None

Close Elasticsearch client connection(s).

If client_key is provided, closes that specific client. Otherwise, closes all registered synchronous clients.

Args:

client_key: Key to identify the synchronous client (default: ‘default’).

async es_query_gen.close_es_client_async(client_key: str | None = 'default') → None

Close asynchronous Elasticsearch client connection(s).

If client_key is provided, closes that specific async client. Otherwise, closes all registered asynchronous clients.

Args:

client_key: Key to identify the asynchronous client (default: ‘default’).

es_query_gen.connect_es(connection_string: str | None = None, host: str = 'localhost', port: int = 9200, scheme: str = 'http', username: str | None = None, password: str | None = None, verify_certs: bool = True, client_key: str = 'default', **kwargs: Any) → Elasticsearch

Create and return a synchronous Elasticsearch client and register it as default.

Args:

connection_string: Full connection string (takes precedence over other params). host: Elasticsearch host (default: ‘localhost’). port: Elasticsearch port (default: 9200). scheme: Connection scheme, ‘http’ or ‘https’ (default: ‘http’). username: Username for authentication (optional). password: Password for authentication (optional). verify_certs: Whether to verify SSL certificates (default: True). client_key: Key to identify the synchronous client (default: ‘default’). kwargs: Additional arguments to pass to Elasticsearch client.

Returns:

The created Elasticsearch client instance.

async es_query_gen.connect_es_async(connection_string: str | None = None, host: str = 'localhost', port: int = 9200, scheme: str = 'http', username: str | None = None, password: str | None = None, verify_certs: bool = True, client_key: str = 'default', **kwargs: Any) → AsyncElasticsearch

Create and return an asynchronous Elasticsearch client and register it as default.

Args:

connection_string: Full connection string (takes precedence over other params). host: Elasticsearch host (default: ‘localhost’). port: Elasticsearch port (default: 9200). scheme: Connection scheme, ‘http’ or ‘https’ (default: ‘http’). username: Username for authentication (optional). password: Password for authentication (optional). verify_certs: Whether to verify SSL certificates (default: True). kwargs: Additional arguments to pass to AsyncElasticsearch client.

Returns:

The created AsyncElasticsearch client instance.

Execute a search query against Elasticsearch with retry and timeout support.

Args:

es: Elasticsearch client (injected by @requires_es_client or passed explicitly). index: Index name or pattern (default: “*” for all indices). query: Elasticsearch query dict (default: empty match_all query). from_: Offset for pagination (default: 0). timeout: Server-side timeout (default: “10s”). max_retries: Number of retry attempts (default: 3). retry_delay: Initial delay between retries in seconds (default: 0.5, with exponential backoff).

Returns:

The full Elasticsearch search response dict.

Raises:

NotFoundError: If the index does not exist. BadRequestError: If the query is malformed. ESTimeoutError: If the server or client timeout is exceeded after retries. ESConnectionError: If unable to connect to Elasticsearch after retries. RequestError: For other Elasticsearch request errors. RuntimeError: If no client is available.

async es_query_gen.es_search_async(es: AsyncElasticsearch | None = None, index: str = '*', query: Dict[str, Any] | None = None, from_: int = 0, timeout: int = 10, max_retries: int = 3, retry_delay: float = 0.5) → Dict[str, Any]

Execute a search query against Elasticsearch (async) with retry and timeout support.

Args:

es: Async Elasticsearch client (injected by @requires_es_client_async or passed explicitly). index: Index name or pattern (default: “*” for all indices). query: Elasticsearch query dict (default: empty match_all query). from_: Offset for pagination (default: 0). timeout: Server-side timeout in seconds (default: 10). max_retries: Number of retry attempts (default: 3). retry_delay: Initial delay between retries in seconds (default: 0.5, with exponential backoff).

Returns:

The full Elasticsearch search response dict.

Raises:

NotFoundError: If the index does not exist. BadRequestError: If the query is malformed. ESTimeoutError: If the server or client timeout is exceeded after retries. ESConnectionError: If unable to connect to Elasticsearch after retries. RequestError: For other Elasticsearch request errors. RuntimeError: If no async client is available.

es_query_gen.get_es_client(client_key: str = 'default') → Elasticsearch | None

Get the module-level default synchronous Elasticsearch client.

Args:

client_key: Key to identify the synchronous client (default: ‘default’).

es_query_gen.get_es_client_async(client_key: str = 'default') → AsyncElasticsearch | None

Get the module-level default asynchronous Elasticsearch client.

Args:

client_key: Key to identify the asynchronous client (default: ‘default’).

es_query_gen.get_es_version(es: Elasticsearch | None = None) → str | None

Return the Elasticsearch version string.

Args:

es: Elasticsearch client (auto-injected if not provided).

Returns:

Version string (e.g., ‘7.10.2’) or None if unavailable.

es_query_gen.get_index_schema(index: str, es: Elasticsearch | None = None) → Dict[str, Any]

Return the mapping/schema for the given index.

Args:

es: Elasticsearch client (auto-injected if not provided). index: The name of the index or alias to retrieve the schema for.

Returns:

Dictionary containing the index mapping/schema keyed by index name.

Raises:

NotFoundError: If the index does not exist.

async es_query_gen.get_index_schema_async(es: AsyncElasticsearch | None = None, index: str = '') → Dict[str, Any]

Return the mapping/schema for the given index asynchronously.

Args:

es: AsyncElasticsearch client (auto-injected if not provided). index: The name of the index or alias to retrieve the schema for.

Returns:

Dictionary containing the index mapping/schema keyed by index name.

Raises:

NotFoundError: If the index does not exist.

es_query_gen.get_index_settings(index: str, es: Elasticsearch | None = None) → Dict[str, Any]

Return the settings for the given index.

Args:

es: Elasticsearch client (auto-injected if not provided). index: The name of the index to retrieve the settings for.

Returns:

Dictionary containing the index settings.

Raises:

NotFoundError: If the index does not exist.

async es_query_gen.get_index_settings_async(es: AsyncElasticsearch | None = None, index: str = '') → Dict[str, Any]

Return the settings for the given index asynchronously.

Args:

es: AsyncElasticsearch client (auto-injected if not provided). index: The name of the index to retrieve the settings for.

Returns:

Dictionary containing the index settings.

Raises:

NotFoundError: If the index does not exist.

es_query_gen.ping(es: Elasticsearch | None = None) → bool

Ping the Elasticsearch cluster to check connectivity.

Args:

es: Elasticsearch client (auto-injected if not provided).

Returns:

True if the cluster is reachable, False otherwise.

async es_query_gen.ping_async(es: AsyncElasticsearch | None = None) → bool

Ping the Elasticsearch cluster asynchronously to check connectivity.

Args:

es: AsyncElasticsearch client (auto-injected if not provided).

Returns:

True if the cluster is reachable, False otherwise.

es_query_gen.search_paginated(es: Elasticsearch | None = None, index: str = '*', query: Dict[str, Any] | None = None, page_size: int = 100, max_docs: int | None = None, timeout: int = 10, max_retries: int = 3, retry_delay: float = 0.5)

Paginate through Elasticsearch results, yielding one full response dict per page.

Repeatedly calls es_search with an increasing from_ offset until there are no more results, the last page is smaller than page_size, or the max_docs limit is reached. Each yielded value is the raw ES response dict (same shape as es_search returns), so you can pass each page directly to ESResponseParser.

Args:

es: Elasticsearch client (auto-injected if not provided). index: Index name or pattern (default: "*"). query: Elasticsearch query dict (default: match_all). page_size: Number of documents per page (default: 100). max_docs: Stop after yielding this many documents in total (default: no limit). timeout: Server-side timeout per request in seconds (default: 10). max_retries: Retry attempts per page request (default: 3). retry_delay: Initial retry backoff in seconds (default: 0.5).

Yields:

Raw Elasticsearch search response dicts, one per page.

Example:

for page in search_paginated(index="my_index", query=my_query, page_size=200):
    docs = ESResponseParser(config).parse_data(page)
    process(docs)
async es_query_gen.search_paginated_async(es: AsyncElasticsearch | None = None, index: str = '*', query: Dict[str, Any] | None = None, page_size: int = 100, max_docs: int | None = None, timeout: int = 10, max_retries: int = 3, retry_delay: float = 0.5)

Async version of search_paginated — an async generator yielding one page per iteration.

Args:

es: AsyncElasticsearch client (falls back to the registered default if not provided). index: Index name or pattern (default: "*"). query: Elasticsearch query dict (default: match_all). page_size: Number of documents per page (default: 100). max_docs: Stop after yielding this many documents in total (default: no limit). timeout: Server-side timeout per request in seconds (default: 10). max_retries: Retry attempts per page request (default: 3). retry_delay: Initial retry backoff in seconds (default: 0.5).

Yields:

Raw Elasticsearch search response dicts, one per page.

Example:

async for page in search_paginated_async(index="my_index", query=my_query):
    docs = ESResponseParser(config).parse_data(page)
    await process(docs)
es_query_gen.set_es_client(es: Elasticsearch, client_key: str = 'default') → None

Set the module-level default synchronous Elasticsearch client.

Args:

es: The Elasticsearch client instance to set as default. client_key: Key to identify the synchronous client (default: ‘default’).

es_query_gen.set_es_client_async(es: AsyncElasticsearch, client_key: str = 'default') → None

Set the module-level default asynchronous Elasticsearch client.

Args:

es: The AsyncElasticsearch client instance to set as default. client_key: Key to identify the asynchronous client (default: ‘default’).

es_query_gen.validate_index(index: str, expected_schema: Dict[str, Any], es: Elasticsearch | None = None, strict_mode: bool = True, ignore_extra_fields: bool = False) → Dict[str, SchemaValidationResult]

Convenience function to validate an Elasticsearch index.

Args:

index (str): Name of the index to validate. expected_schema (Dict[str, Any]): The expected schema configuration. es (Optional[Elasticsearch]): Elasticsearch client (auto-injected if not provided). strict_mode (bool): If True, treat all mismatches as errors. ignore_extra_fields (bool): If True, don’t report extra fields.

Returns:

Dict[str, SchemaValidationResult]: Validation results keyed by index name.

Example:

from es_query_gen import connect_es

client = connect_es(host='localhost')
expected = {
    "mappings": {
        "properties": {
            "name": {"type": "text"},
            "age": {"type": "integer"}
        }
    }
}
results = validate_index("my_index", expected, es=client)
for index_name, result in results.items():
    if not result.is_valid:
        print(index_name, result)