es-query-gen 1.15.0

Library to build complex ES queries from simple Python dictionaries using typed Pydantic models.

es-query-gen

Documentation

A no-code Elasticsearch query generator - Build complex ES queries from simple Python dictionaries using typed Pydantic models.

Overview

This library provides five main components:

1. Query Generator - Convert simple configuration dictionaries into complex Elasticsearch DSL queries
2. ES Client Wrapper - Simplified connection management with retry logic, timeouts, and both sync/async support
3. Response Parser - Parse complex Elasticsearch responses including nested aggregations into clean Python objects
4. Response Transformer - Transform parsed documents (type casting, renaming, casing, date conversion)
5. Schema Validator - Validate Elasticsearch index schemas to ensure settings and field mappings match expected configurations

Notes

• Helper rename: The module-level client helpers were renamed to set_es_client, get_es_client, and clear_es_client for synchronous clients, with async equivalents set_es_client_async, get_es_client_async, and clear_es_client_async.
• Client registry: The client registry supports storing clients by key. A registry entry may hold either a synchronous Elasticsearch or an AsyncElasticsearch instance; use the matching helper or async-prefixed methods when interacting with the client.
• Async methods disclaimer: Async helpers and operations use an _async suffix (for example connect_es_async, ping_async, es_search_async, get_index_schema_async). Always await these async-prefixed functions and do not call them from synchronous code.

Features

• 🎯 No-code query building - Define queries using simple Python dicts or JSON
• 📝 Typed models - Full Pydantic validation for filters, aggregations, and configurations
• 🔄 Query Builder - Convert models into Elasticsearch DSL with support for:
  • Equality and inequality filters
  • Numeric and date range filters (with relative date offsets)
  • Sorting and pagination
  • Nested aggregations (unlimited depth)
  • Field selection
• 🔌 ES Client Management - Singleton pattern with connection pooling, retries, and error handling
• 📊 Response Parser - Extract documents from complex nested aggregations
• 🛠️ Response Transformer - Transform output fields (rename, auto-convert datetimes to epoch, uppercase/lowercase, default values)
• ⚡ Async Support - Full async/await support for all ES operations (connect_es_async, es_search_async, get_index_schema_async, etc.)
• 📝 Comprehensive Logging - Built-in logging with performance metrics, compatible with JSON/structured logging
• ✅ Schema Validator - Validate index schemas against expected configurations with detailed reporting
• ✅ 100+ Tests - Comprehensive test coverage with fixtures and integration tests

Requirements

• Python 3.10+ (uses match statement)
• Pydantic 2.x
• Elasticsearch Python client 9.x

Installation

Using pip

Install from source in editable mode:

pip install -e .

Using uv (recommended)

uv is a fast Python package installer:

# Install uv if you haven't already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install the package
uv pip install -e .

# Or install with development dependencies
uv pip install -e ".[dev]"

Quick Start

1. Build a Query (No Code!)

Define your query using a simple Python dictionary:

from es_query_gen import QueryBuilder

# Define query using a simple dict - no ES DSL knowledge needed!
config = {
    "searchFilters": {
        "equals": [{"field": "status", "value": "active"}],
        "rangeFilters": [{"field": "age", "gte": 18, "lte": 65, "rangeType": "number"}]
    },
    "sortList": [{"field": "created_at", "order": "desc"}],
    "size": 10,
    "returnFields": ["id", "name", "email"]
}

# Build ES query automatically
query = QueryBuilder().build(config)
print(query)
# Output: Full Elasticsearch DSL query ready to execute!

2. Connect to Elasticsearch

Synchronous:

from es_query_gen import connect_es, es_search

# Connect with automatic client management
client = connect_es(
    host='localhost',
    username='elastic',
    password='changeme',
    request_timeout=30
)

# Execute the query with retry logic
response = es_search(index='my_index', query=query)

Asynchronous:

import asyncio
from es_query_gen import connect_es_async, es_search_async

async def main():
    # Connect with async client
    client = await connect_es_async(
        host='localhost',
        username='elastic',
        password='changeme',
        request_timeout=30
    )
    
    # Execute the query asynchronously
    response = await es_search_async(index='my_index', query=query)
    return response

# Run async code
response = asyncio.run(main())

3. Parse Complex Responses

from es_query_gen import ESResponseParser

# Parse search results or complex nested aggregations
parser = ESResponseParser(config)
results = parser.parse_data(response)

# Results are clean Python dicts, even from nested aggregations!
for doc in results:
    print(doc['name'], doc['email'])

4. Transform Output Data

Clean and reshape the parsed documents before returning them to your application using ResponseTransformer:

from es_query_gen import TransformConfig, FieldTransformRule, ResponseTransformer

# Define how you want your parsed data mapped, renamed, or cast
config = TransformConfig(
    rules=[
        # Rename and cast types:
        FieldTransformRule(source_key="user_id", target_key="userId", type_cast="STR"),
        
        # Apply default values for missing keys:
        FieldTransformRule(source_key="score", type_cast="FLOAT", default=0.0),
        
        # Auto-convert datetime objects or ISO strings into UTC Unix epochs:
        FieldTransformRule(source_key="created_at", target_key="createdAt", type_cast="EPOCH"),
        
        # Convert datetime objects/strings using explicit date formats:
        FieldTransformRule(
            source_key="birthday", 
            type_cast="DATE", 
            date_format="%d-%m-%Y",      # how to parse the string (defaults to ISO)
            target_date_format="%Y-%m-%d" # how to output the string
        ),
        
        # Apply text casing (UPPER, LOWER, TITLE):
        FieldTransformRule(source_key="status", value_casing="UPPER"),
    ],
    # Whether to keep fields not listed in rules above (default: True)
    include_unmapped=True
)

transformer = ResponseTransformer(config)

# Transform the previously parsed `results`
transformed_results = transformer.transform(results)

5. Configure Logging

The library uses Python's standard logging module and automatically inherits the logging configuration from your application:

import logging
from es_query_gen import connect_es, es_search

# Configure logging for your application
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# The library will use your logger configuration
client = connect_es(host='localhost', username='elastic', password='changeme')
response = es_search(index='my_index', query=query)

# Logs will show:
# 2026-01-17 10:30:45 - es_query_gen.es_utils.connection - INFO - Ping completed in 15.23ms
# 2026-01-17 10:30:46 - es_query_gen.es_utils.connection - INFO - Search completed in 142.56ms (index='my_index', from=0, query={...})

JSON Logging:

The library works seamlessly with JSON logging libraries:

import logging
from pythonjsonlogger import jsonlogger
from es_query_gen import connect_es, es_search

# Configure JSON logger
logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    '%(asctime)s %(name)s %(levelname)s %(message)s'
)
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# Library logs will automatically be in JSON format
client = connect_es(host='localhost')
response = es_search(index='my_index', query=query)

# Output: {"asctime": "2026-01-17 10:30:45", "name": "es_query_gen.es_utils.connection", 
#          "levelname": "INFO", "message": "Search completed in 142.56ms..."}

What Gets Logged:

• Performance metrics: Timing for all ES operations (ping, search) in milliseconds
• Query details: Index name, query parameters, pagination (from, size)
• Connection events: Client initialization, connection status
• Validation results: Schema validation errors and warnings
• Retry attempts: Timeout/connection errors with retry counts

See LOGGING.md for comprehensive logging documentation and examples.

5. Validate Index Schemas

Ensure your Elasticsearch indices match expected configurations:

from es_query_gen import validate_index, SchemaValidator

# Define expected schema
expected_schema = {
    "settings": {
        "number_of_shards": 1,
        "analysis": {
            "normalizer": {
                "lowercase_normalizer": {
                    "type": "custom",
                    "filter": ["lowercase"]
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "name": {
                "type": "text",
                "fields": {
                    "keyword": {"type": "keyword", "normalizer": "lowercase_normalizer"}
                }
            },
            "age": {"type": "integer"},
            "email": {"type": "keyword"}
        }
    }
}

# Validate against live index or alias
results = validate_index("my_index_or_alias", expected_schema, es=client)

for index_name, result in results.items():
    if not result.is_valid:
        print(index_name, result)  # Detailed error report per index
        # Output:
        # Schema Validation: FAILED
        #
        # Errors (2):
        #   - Missing field in schema: email
        #   - Type mismatch for field 'age': expected 'integer', got 'text'
        #
        # Missing Fields (1):
        #   - email

# Use custom validator with options
validator = SchemaValidator(strict_mode=False, ignore_extra_fields=True)
result = validator.validate_index("my_index", expected_schema)

# Check specific validation details
if result.type_mismatches:
    for field, (expected, actual) in result.type_mismatches.items():
        print(f"Field {field}: expected {expected}, got {actual}")

Schema Validator Features:

• ✅ Settings validation - Compare index settings (shards, replicas, analyzers, normalizers)
• ✅ Field type validation - Ensure field types match (text, keyword, integer, date, etc.)
• ✅ Field properties validation - Validate format, analyzer, normalizer, and other field settings
• ✅ Nested fields support - Handle multi-fields and nested object properties
• ✅ Detailed reporting - Get comprehensive results with errors, warnings, and specific mismatches
• ✅ Flexible modes - Strict mode for errors, non-strict for warnings, ignore extra fields option
• ✅ Convenience functions - Simple one-line validation with validate_index() or validate_schema()

Advanced Example: Nested Aggregations

Build complex aggregation queries without writing ES DSL:

{
    "size": 10,
    "searchFilters": {
        "equals": [
            {
                "field": "age",
                "value": "35"
            }
        ],
        "rangeFilters": [
            {
                "field": "dob",
                "rangeType": "date",
                "dateFormat": "%m/%d/%Y",
                "gte": {
                    "month": 2,
                    "years": -60
                },
                "lt": {
                    "month": 9,
                    "day": 10,
                    "years": -20
                }
            }
        ]
    },
    "sortList": [
        {
            "field": "dob",
            "order": "asc"
        }
    ],
    "returnFields": ["name", "dob", "phone"],
    "aggs": [
        {
            "name": "address_bucket",
            "field": "address.keyword",
            "size": 100,
            "order": "asc"
        },
        {
            "name": "dob_bucket",
            "field": "dob",
            "size": 100,
            "order": "asc"
        }
    ]
}

Testing

This library includes a comprehensive test suite with 100+ tests covering all components.

Quick Start

Run all unit tests (excludes integration tests that require ES):

# Using pip
pip install -e ".[dev]"
pytest -m "not integration"

# Using uv (faster)
uv pip install -e ".[dev]"
pytest -m "not integration"

# Run with coverage report
pytest -m "not integration" --cov=src/es_query_gen --cov-report=html

Or use the provided test runner script:

./run_tests.sh

Test Structure

• tests/test_models.py - Tests for Pydantic models and validators (50+ tests)
• tests/test_builder.py - Tests for QueryBuilder query construction (40+ tests)
• tests/test_parser.py - Tests for ESResponseParser (35+ tests)
• tests/test_connection.py - Tests for ES connection management (40+ tests)
• tests/test_schema_validator.py - Tests for schema validation (40+ tests)
• tests/test_integration.py - Integration tests (requires running ES)
• tests/conftest.py - Shared fixtures and test configuration

Integration Tests

Integration tests require a running Elasticsearch instance:

# Start local ES (using provided docker-compose)
cd elastic-start-local && ./start.sh

# Configure connection (optional, defaults shown)
export ES_HOST=localhost
export ES_PORT=9200
export ES_USERNAME=elastic
export ES_PASSWORD=changeme

# Run integration tests
pytest -m integration

Coverage

The test suite provides comprehensive coverage:

• Models: 100% - All validation logic and edge cases
• Builder: 100% - All query building paths
• Parser: 100% - Search and aggregation parsing
• Connection: High - All connection and operation flows
• Schema Validator: High - All validation scenarios and edge cases

View detailed coverage report:

pytest --cov=src/es_query_gen --cov-report=html
open htmlcov/index.html

See tests/README.md for detailed testing documentation.

API Components

QueryBuilder

• Converts configuration dicts to ES DSL
• Supports filters, sorting, pagination, aggregations
• Handles date math and relative date ranges

ES Client Wrapper

• Singleton pattern for connection management
• Automatic retry with exponential backoff
• Both sync and async support:
  • Sync: connect_es(), es_search(), get_index_schema(), get_index_settings()
  • Async: connect_es_async(), es_search_async(), get_index_schema_async(), get_index_settings_async()
• Decorators for client injection (@requires_es_client, @requires_es_client_async)
• Built-in logging with configurable handlers

Response Parser

• Extracts documents from search results
• Parses nested aggregations (any de

Schema Validator

• Validates index schemas against expected configurations
• Compares settings and field mappings
• Supports strict and non-strict validation modes
• Detailed error and warning reporting
• Handles nested fields and complex structurespth)
• Preserves all _source fields + _id

Documentation (Sphinx)

How to build and update the Sphinx documentation locally.

1. Install development dependencies (includes Sphinx):

pip install -e .[dev]

2. Build static HTML docs:

# Build into docs/_build/html
sphinx-build -b html docs/source docs/_build/html

# Open the generated docs (macOS)
open docs/_build/html/index.html

If you prefer the Makefile helpers (common in Sphinx projects), you can use:

# From the repo root (when a Makefile exists under `docs`)
make -C docs clean
make -C docs html

# or, from inside the docs directory if a Makefile is present
cd docs && make clean && make html

# The above `make html` will place built files in `docs/_build/html`

If you don't have a Makefile, sphinx-build shown above is the equivalent.

3. Live-reload while editing docs:

# Requires sphinx-autobuild (in dev extras)
sphinx-autobuild docs/source docs/_build/html

4. Notes on API autodoc

• We enabled sphinx.ext.autodoc and the src directory is added to sys.path in docs/source/conf.py so .. automodule:: es_query_gen will document the package exports.
• If you add new public functions or classes to the package, update their docstrings and then rebuild the docs to include them.

5. Commit changes

• The built HTML files are under docs/_build/html. You generally should not commit built HTML unless you host them in the repo (e.g., GitHub Pages). Commit source files under docs/source.

Contributing

Contributions are welcome! Here's how to get started:

Setup Development Environment

# Clone the repository
git clone https://github.com/yourusername/es-query-gen.git
cd es-query-gen

# Using uv (recommended - much faster)
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -e ".[dev]"

# Or using pip
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Run Tests

# Run unit tests
pytest -m "not integration"

# Run with coverage
pytest --cov=src/es_query_gen --cov-report=html

# Run all tests including integration (requires ES)
pytest

Development Guidelines

1. Keep changes small - Focus on one feature/fix per PR
2. Add tests - All new features must include tests
3. Follow existing patterns - Match the code style
4. Update documentation - Keep README and docstrings current
5. Type hints - Use Pydantic models and type annotations

Making Changes

# Create a feature branch
git checkout -b feature/your-feature-name

# Make your changes and add tests
# ...

# Run tests
pytest

# Commit with clear message
git commit -m "Add feature: description"

# Push and create PR
git push origin feature/your-feature-name

License

MIT

Notes

• Requires Python 3.10+ (uses match statement)
• The library is designed to be extended - you can add custom query builders or parsers
• For production use, consider implementing connection pooling based on your needs