# Elasticsearch Sdk > Integrates Elasticsearch using elasticsearch-py 8.x with patterns for indexing, search queries, aggregations, vector search, bulk operations, and index lifecycle management. - Skill: `paulpas/elasticsearch-sdk` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/elasticsearch-sdk` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/elasticsearch-sdk/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/elasticsearch-sdk --- # Elasticsearch Python SDK Integration Integrates Elasticsearch using `elasticsearch-py` 8.x (also compatible with OpenSearch) with patterns for index management, search queries (term, match, bool), aggregations, bulk indexing, vector/kNN search, index lifecycle management (ILM), and async operations. ## TL;DR Checklist - [ ] Use `Elasticsearch()` client with `basic_auth` or `api_key` for authentication - [ ] Use `helpers.bulk()` for high-throughput indexing — never index documents one-by-one - [ ] Use `bool` query with `must`/`filter`/`should` clauses for complex search - [ ] Use `aggs` for bucket and metric aggregations, not client-side grouping - [ ] Use `knn` query for vector similarity search (Elasticsearch 8.x+) - [ ] Use `index.create()` with explicit mappings — never rely on dynamic mapping for production - [ ] Use `point_in_time` (PIT) + `search_after` for deep pagination instead of `from`/`size` --- ## When to Use Use this skill when: - Building full-text search functionality for applications - Implementing log aggregation and analysis with Elasticsearch - Performing real-time analytics with bucket and metric aggregations - Implementing vector/kNN search for semantic or similarity-based retrieval - Building RAG (Retrieval-Augmented Generation) pipelines with Elasticsearch as a vector store - Managing time-series indices with ILM policies and rollover - Migrating from or integrating with OpenSearch (API-compatible) --- ## When NOT to Use - For transaction-heavy OLTP workloads (use PostgreSQL or MongoDB) - When you need a primary data store with strong consistency guarantees - For simple key-value lookups (use Redis or DynamoDB) - For complex joins across multiple entities (use a relational database) - When your data fits in memory and doesn't need search capabilities (keep it simple) --- ## Core Workflow ### 1. Connect to Elasticsearch ```python from elasticsearch import Elasticsearch, helpers from elasticsearch.exceptions import ( ConnectionError as ESConnectionError, NotFoundError, AuthorizationException, ) # Elastic Cloud or self-managed cluster es = Elasticsearch( "https://localhost:9200", basic_auth=("elastic", os.environ["ES_PASSWORD"]), ca_certs="/path/to/ca.crt", request_timeout=30, max_retries=3, retry_on_timeout=True, ) # Verify connectivity if not es.ping(): raise RuntimeError("Cannot connect to Elasticsearch") ``` **Checkpoint:** Always verify with `es.ping()` or `es.info()` at startup. Catch `AuthorizationException` immediately on auth failures — don't proceed with invalid credentials. ### 2. Create Index with Explicit Mappings ```python def create_articles_index(es: Elasticsearch, index_name: str) -> dict: """Create an index with explicit mappings and settings.""" mappings = { "properties": { "title": { "type": "text", "fields": {"keyword": {"type": "keyword"}}, "analyzer": "english", }, "content": {"type": "text", "analyzer": "english"}, "author": {"type": "keyword"}, "published_at": {"type": "date"}, "tags": {"type": "keyword"}, "view_count": {"type": "integer"}, "embedding": { "type": "dense_vector", "dims": 768, "index": True, "similarity": "cosine", }, } } settings = { "number_of_shards": 3, "number_of_replicas": 1, "refresh_interval": "30s", } if not es.indices.exists(index=index_name): return es.indices.create(index=index_name, mappings=mappings, settings=settings) return {"acknowledged": True, "existing": True} ``` **Checkpoint:** Always define explicit mappings before indexing production data. Use `keyword` type for exact-match fields and `text` with analyzers for full-text search. Add `dense_vector` fields for vector search. ### 3. Index Documents with Bulk API ```python def bulk_index_articles(es: Elasticsearch, index_name: str, articles: list[dict]) -> dict: """Bulk index articles with error handling.""" def generate_actions(): for article in articles: yield { "_index": index_name, "_id": article.get("id"), "_source": { "title": article["title"], "content": article["content"], "author": article["author"], "published_at": article["published_at"], "tags": article.get("tags", []), "view_count": article.get("view_count", 0), "embedding": article.get("embedding"), }, } success, errors = helpers.bulk( es, generate_actions(), chunk_size=500, request_timeout=60, raise_on_error=False, ) return {"indexed": success, "errors": len(errors)} ``` **Checkpoint:** Use `raise_on_error=False` and inspect errors for failures. Use `chunk_size` to balance throughput and memory. Always check `errors` count after bulk operations. ### 4. Search with Bool Query ```python def search_articles( es: Elasticsearch, index_name: str, query_text: str, author: str | None = None, tags: list[str] | None = None, min_date: str | None = None, from_: int = 0, size: int = 20, ) -> dict: """Full-text search with optional filters.""" must_clauses = [{"match": {"content": query_text}}] filter_clauses = [] if author: filter_clauses.append({"term": {"author": author}}) if tags: filter_clauses.append({"terms": {"tags": tags}}) if min_date: filter_clauses.append({"range": {"published_at": {"gte": min_date}}}) body = { "query": { "bool": { "must": must_clauses, "filter": filter_clauses, } }, "sort": [{"_score": "desc"}, {"published_at": "desc"}], "from_": from_, "size": size, "aggs": { "by_author": {"terms": {"field": "author", "size": 10}}, "date_histogram": {"date_histogram": {"field": "published_at", "calendar_interval": "month"}}, }, } return es.search(index=index_name, body=body) ``` **Checkpoint:** Use `filter` context for structured data (dates, keywords) — filter clauses are cached and don't affect scoring. Use `must` for full-text relevance scoring. --- ## Implementation Patterns ### Pattern 1: Vector/kNN Search ```python def vector_search( es: Elasticsearch, index_name: str, query_vector: list[float], k: int = 10, num_candidates: int = 100, ) -> list[dict]: """Perform kNN vector search for semantic similarity.""" body = { "knn": { "field": "embedding", "query_vector": query_vector, "k": k, "num_candidates": num_candidates, }, "_source": {"excludes": ["embedding"]}, } response = es.search(index=index_name, body=body) return [hit["_source"] for hit in response["hits"]["hits"]] ``` ### Pattern 2: Point-in-Time Deep Pagination ```python def scroll_all_results(es: Elasticsearch, index_name: str, query: dict, page_size: int = 1000): """Paginate through all matching results using PIT + search_after.""" pit = es.open_point_in_time(index=index_name, keep_alive="5m") sort = [{"published_at": "desc"}, {"_shard_doc": "desc"}] last_sort = None try: while True: body = { "size": page_size, "query": query, "sort": sort, "pit": {"id": pit["id"], "keep_alive": "5m"}, } if last_sort: body["search_after"] = last_sort response = es.search(body=body) hits = response["hits"]["hits"] if not hits: break for hit in hits: yield hit["_source"] last_sort = hits[-1]["sort"] finally: es.close_point_in_time(id=pit["id"]) ``` ### BAD vs GOOD: Indexing Pattern ```python # ❌ BAD — Indexing one document at a time (slow) def index_articles_bad(es, index, articles): for article in articles: es.index(index=index, document=article) # ✅ GOOD — Bulk indexing (10-50x faster) def index_articles_good(es, index, articles): success, errors = helpers.bulk(es, ( {"_index": index, "_source": a} for a in articles ), raise_on_error=False) if errors: log.error(f"Bulk indexing failed for {len(errors)} documents") ``` ### BAD vs GOOD: Query Construction ```python # ❌ BAD — Inefficient wildcard query def search_bad(es, index, term): return es.search(index=index, query={"wildcard": {"title": f"*{term}*"}}) # ✅ GOOD — Analyzed match query with proper field def search_good(es, index, term): return es.search(index=index, query={"match": {"title": {"query": term, "fuzziness": "AUTO"}}}) ``` --- ## Constraints ### MUST DO - Define explicit index mappings before indexing — never rely solely on dynamic mapping - Use `helpers.bulk()` for all batch indexing operations - Use bool queries with `filter` for structured data and `must` for full-text relevance - Set `number_of_shards` based on your data volume and node count (20-40GB per shard) - Use `ilm` (Index Lifecycle Management) for time-series indices - Set `refresh_interval` to `-1` during bulk indexing, reset after ### MUST NOT DO - Never use `scroll` for deep pagination in user-facing queries — use `search_after` or PIT - Avoid `wildcard` and `regexp` queries on `text` fields — they skip the analysis index - Do not index documents one-by-one in loops — always use bulk helpers - Never use `match_all` queries without `size` limits — can crash the cluster - Avoid large `terms` queries with thousands of values — use `terms_set` or filters instead --- ## Output Template When writing Elasticsearch integration code, structure your output as: 1. **Client Initialization** — Elasticsearch client with authentication and connection settings 2. **Index Setup** — Create index with explicit mappings (fields, analyzers, vector config) 3. **Bulk Indexing** — helpers.bulk() with chunked document generation 4. **Search Query** — Bool query with must/filter/should clauses, aggregations, sorting 5. **Result Processing** — Extract hits, parse aggregations, handle pagination --- ## Related Skills | Skill | Purpose | |---|---| | `coding-mongodb-driver` | Document database patterns — alternative to Elasticsearch for non-search workloads | | `coding-rag-architecture` | RAG pipeline patterns using Elasticsearch as vector store | | `coding-postgresql-sdk` | PostgreSQL full-text search as an alternative | --- ## Live References - [elasticsearch-py Documentation](https://elasticsearch-py.readthedocs.io/en/v8.x/) — Official Python client 8.x docs - [Elasticsearch Query DSL](https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl.html) — Complete query DSL reference - [Elasticsearch Aggregations Guide](https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations.html) — Bucket and metric aggregations - [Elasticsearch Vector Search](https://www.elastic.co/guide/en/elasticsearch/reference/current/knn-search.html) — kNN and vector similarity search - [Elasticsearch Bulk API](https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html) — Bulk indexing and update API - [Elasticsearch ILM](https://www.elastic.co/guide/en/elasticsearch/reference/current/index-lifecycle-management.html) — Index lifecycle management policies - [OpenSearch Python Client](https://opensearch.org/docs/latest/clients/python/) — OpenSearch Python client (API-compatible)