How to use Elasticsearch in Python

Elasticsearch is a powerful search and analytics engine for modern applications. Its integration with Python allows developers to build sophisticated, data-driven features with speed and efficiency.

In this article, we'll cover key techniques and practical tips. We will show you real-world applications and provide effective advice to debug issues as you master Elasticsearch in Python.

Basic connection to Elasticsearch

from elasticsearch import Elasticsearch

# Connect to Elasticsearch
es = Elasticsearch("http://localhost:9200")
print(f"Connected: {es.ping()}")--OUTPUT--Connected: True

To begin interacting with your cluster, you first create a client instance from the Elasticsearch class. The code initializes this connection using the default endpoint for a local Elasticsearch server, http://localhost:9200. This es object becomes your primary tool for all subsequent operations.

While creating the object establishes a potential connection, the es.ping() method actively verifies it. This isn't just a simple network check; it confirms two key things:

• The client can successfully reach the Elasticsearch cluster.
• The cluster is alive and ready to process requests.

Receiving a True response is a reliable handshake, confirming you're ready to start indexing and searching data.

Basic Elasticsearch operations

With your connection confirmed, you can now perform the three core data operations: indexing new documents, retrieving them, and running basic searches.

Indexing a document

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
doc = {"title": "Python Guide", "content": "Elasticsearch tutorial", "tags": ["python", "search"]}
response = es.index(index="articles", document=doc)
print(f"Document indexed with ID: {response['_id']}")--OUTPUT--Document indexed with ID: jK8_RIcBnPskDzOkXkM5

Indexing is how you add data to Elasticsearch, making it searchable. You simply structure your data as a Python dictionary and pass it to the es.index() method. This function handles the entire process of sending and storing your document.

• The index parameter specifies where to store the document, in this case, an index named "articles".
• The document parameter is the Python dictionary you want to add.

Elasticsearch confirms the operation by returning a response that includes a unique _id for the newly created document.

Retrieving a document with get()

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
document_id = "jK8_RIcBnPskDzOkXkM5"
result = es.get(index="articles", id=document_id)
print(f"Found document: {result['_source']}")--OUTPUT--Found document: {'title': 'Python Guide', 'content': 'Elasticsearch tutorial', 'tags': ['python', 'search']}

When you know a document's unique ID, the es.get() method is the most direct way to fetch it. It’s a targeted lookup, not a broad search. You simply provide two key pieces of information:

• The index where the document is stored.
• The specific id of the document you want.

Elasticsearch returns a dictionary, and your original data is neatly tucked inside the _source key.

Basic search with search()

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
query = {"query": {"match": {"content": "elasticsearch"}}}
results = es.search(index="articles", body=query)
print(f"Hits: {len(results['hits']['hits'])}")
print(f"First hit: {results['hits']['hits'][0]['_source']}")--OUTPUT--Hits: 1
First hit: {'title': 'Python Guide', 'content': 'Elasticsearch tutorial', 'tags': ['python', 'search']}

The es.search() method is your primary tool for finding documents that match specific criteria. Unlike get(), it performs a broad search based on a query you define. The query itself is a dictionary. In this example, a match query looks for the term "elasticsearch" within the content field.

• Elasticsearch returns a detailed response, with the actual documents found in a nested list at results['hits']['hits'].
• Each item in this list represents a single hit, and the original document is available in its _source key.

Advanced Elasticsearch techniques

Moving beyond single-document operations, you're ready to handle data at scale using bulk processing, aggregations, and the powerful Query DSL.

Using bulk operations

from elasticsearch import Elasticsearch, helpers

es = Elasticsearch("http://localhost:9200")
actions = [
{"_index": "books", "_source": {"title": "Book 1", "author": "Author A"}},
{"_index": "books", "_source": {"title": "Book 2", "author": "Author B"}},
{"_index": "books", "_source": {"title": "Book 3", "author": "Author C"}}
]
success, failed = helpers.bulk(es, actions, stats_only=True)
print(f"Successfully indexed: {success}, Failed: {failed}")--OUTPUT--Successfully indexed: 3, Failed: 0

Indexing documents one by one is inefficient for large datasets. The helpers.bulk() function streamlines this process by sending multiple operations in a single network request, which significantly reduces overhead and improves performance.

• You prepare a list of actions, where each dictionary defines an operation like indexing.
• Each action must specify the target _index and the document data in _source.
• The function returns a simple count of successful and failed operations when you set stats_only=True.

Working with aggregations

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
query = {
"aggs": {
"popular_authors": {
"terms": {"field": "author.keyword", "size": 5}
}
},
"size": 0
}
results = es.search(index="books", body=query)
print(f"Top authors: {results['aggregations']['popular_authors']['buckets']}")--OUTPUT--Top authors: [{'key': 'Author A', 'doc_count': 1}, {'key': 'Author B', 'doc_count': 1}, {'key': 'Author C', 'doc_count': 1}]

Aggregations are Elasticsearch's powerful analytics engine. Instead of just fetching documents, you can use them to summarize your data. This query uses a terms aggregation to group all documents in the books index by the unique values in the author.keyword field.

• The aggs object is where you define the specific analysis you want to perform.
• Setting size: 0 is a key optimization. It tells Elasticsearch to return only the aggregation results, not the actual documents, which speeds up the query.
• The results appear in a list of buckets, where each bucket contains a unique author and their corresponding document count.

Using the query DSL for complex searches

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
complex_query = {
"query": {
"bool": {
"must": [{"match": {"title": "Book"}}],
"filter": [{"term": {"author.keyword": "Author B"}}]
}
}
}
results = es.search(index="books", body=complex_query)
print(f"Matching documents: {[hit['_source'] for hit in results['hits']['hits']]}")--OUTPUT--Matching documents: [{'title': 'Book 2', 'author': 'Author B'}]

The Query DSL is your toolkit for building sophisticated searches by combining multiple conditions. This example uses a bool query, which acts as a container for more specific rules.

• The must clause is for queries that have to match and contribute to the relevance score. Here, it finds documents where the title contains "Book".
• The filter clause is for strict, yes-or-no criteria. It doesn't affect scoring, making it very efficient for exact matches like finding where author.keyword is "Author B".

Move faster with Replit

Replit is an AI-powered development platform that comes with all Python dependencies pre-installed, so you can skip setup and start coding instantly. Instead of piecing together techniques, you can describe the app you want to build, and Agent 4 will take it from idea to working product:

• A real-time product search for an e-commerce site, using complex queries to filter by brand and availability.
• An analytics dashboard that uses aggregations to display the most popular authors from a collection of articles.
• A data migration script that efficiently indexes thousands of documents into Elasticsearch using bulk operations.

Simply describe your app, and Replit will write the code, test it, and fix issues automatically, all within your browser.

Common errors and challenges

Navigating Elasticsearch in Python can present a few common challenges, but they're all manageable with the right approach.

Network issues or a downed cluster can interrupt your application, leading to a ConnectionError. Instead of letting your program crash, you can handle these situations gracefully by wrapping your Elasticsearch calls in a try-except block. This allows you to catch the error, log it for debugging, and perhaps retry the connection after a short delay.

Elasticsearch often guesses your data's format, but this dynamic mapping isn't always perfect and can cause search problems. For instance, it might treat a numeric ID as text. You can avoid this by defining an explicit mapping when you create an index, which tells Elasticsearch exactly how to handle each field and ensures your queries behave as expected.

Fetching a massive result set in one go is inefficient and can overload your system. A better strategy is to implement pagination using the from and size parameters in your search query. The size parameter limits how many documents are returned in one batch, while from sets the starting point, allowing you to retrieve results page by page.

Handling connection errors with proper try-except blocks

A ConnectionError is a common roadblock that occurs when your client can't communicate with the Elasticsearch server. This could be due to a network issue or a simple typo in the host address. Without error handling, your application will halt abruptly.

The script below illustrates this problem by attempting to connect to an incorrect hostname, leading to an unhandled error.

from elasticsearch import Elasticsearch

# Connect without error handling
es = Elasticsearch("http://wrong-hostname:9200")
response = es.index(index="test", document={"field": "value"})
print("Document indexed successfully")

The script fails because it points to an invalid server, "wrong-hostname". The connection isn't attempted until es.index() is called, which is when the program crashes. The corrected approach below shows how to anticipate and manage this failure.

from elasticsearch import Elasticsearch
from elasticsearch.exceptions import ConnectionError, ConnectionTimeout

try:
es = Elasticsearch("http://wrong-hostname:9200", timeout=5)
response = es.index(index="test", document={"field": "value"})
print("Document indexed successfully")
except ConnectionError as e:
print(f"Connection error: {e}")
except ConnectionTimeout:
print("Connection timed out")

The solution wraps the connection attempt in a try-except block, which lets your application gracefully handle failures instead of crashing. The except ConnectionError block catches issues like an incorrect hostname, while except ConnectionTimeout handles slow responses. Setting a timeout in the Elasticsearch client is a proactive measure. This pattern is crucial whenever your code interacts with external services, as network reliability is never guaranteed.

Fixing field mapping issues with explicit mappings

When Elasticsearch automatically maps a field like user_id as text, it rejects later documents that try to use a number for that same field. This data type mismatch is a common source of indexing failures. See how it happens in the following code.

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
# First document establishes the field as a string
es.index(index="users", document={"user_id": "12345"})
# Second document tries to use a number for the same field
es.index(index="users", document={"user_id": 67890})

The first document’s string user_id locks in the data type. When the second document uses an integer for the same field, it causes an indexing failure. The corrected approach below prevents this conflict from happening.

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
mapping = {
"mappings": {
"properties": {
"user_id": {"type": "long"}
}
}
}
es.indices.create(index="users", body=mapping, ignore=400)
es.index(index="users", document={"user_id": 12345})
es.index(index="users", document={"user_id": 67890})

The solution is to define an explicit mapping before indexing any documents. By using es.indices.create(), you instruct Elasticsearch to treat the user_id field as a long, which preemptively solves data type conflicts.

The ignore=400 parameter is a practical touch. It tells the client to simply ignore the error if the index already exists, making your script more robust and runnable multiple times without failing.

Implementing pagination for large result sets with from and size

Fetching thousands of documents in a single query is inefficient and can strain both your application and the Elasticsearch cluster. By default, an es.search() query only returns the top 10 results, which can be misleading. The code below demonstrates this limitation.

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
query = {"query": {"match_all": {}}}
# This only returns the default 10 results
results = es.search(index="products", body=query)
all_products = [hit["_source"] for hit in results["hits"]["hits"]]
print(f"Found {len(all_products)} products")

The code executes a match_all query but only processes the default 10 results returned by es.search(). The following example shows how to fetch documents beyond this initial limit.

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
all_products = []
for page in range(5): # Fetch first 5 pages
results = es.search(index="products", body={"query": {"match_all": {}}},
size=100, from_=page*100)
all_products.extend([hit["_source"] for hit in results["hits"]["hits"]])
print(f"Found {len(all_products)} products")

This corrected code fetches results in batches. The size parameter tells es.search() how many documents to return, while from_ sets the starting point for each batch. By looping and increasing the from_ offset, you can efficiently process thousands of documents without overwhelming your application. Use this pagination strategy whenever a search could return a large number of hits, as it keeps your application stable and performant.

Real-world applications

With these foundational skills, you can build practical applications, from e-commerce product searches to time-series analysis with date_histogram.

Building a simple product search for e-commerce

You can build a precise product search by combining a relevance-based match query with a strict filter, ensuring a search for "laptop" only returns results from the "electronics" category.

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")
query = {
"query": {
"bool": {
"must": [{"match": {"name": "laptop"}}],
"filter": [{"term": {"category.keyword": "electronics"}}]
}
}
}
results = es.search(index="products", body=query)
for product in results["hits"]["hits"]:
print(f"{product['_source']['name']} - ${product['_source']['price']}")

This code executes a layered search using a bool query to find specific products. It combines two distinct conditions to refine the results:

• The must clause finds items where the name field contains "laptop."
• The filter clause then narrows those results, keeping only items where the category.keyword field is an exact match for "electronics."

After the search runs, the code loops through the returned hits, extracts the original product data from _source, and prints the name and price of each match.

Analyzing time-series data with date_histogram

You can use a date_histogram aggregation to analyze time-series data, grouping documents into fixed intervals like days or hours to track trends such as daily error counts.

from elasticsearch import Elasticsearch
from datetime import datetime, timedelta

es = Elasticsearch("http://localhost:9200")
now = datetime.now()
week_ago = now - timedelta(days=7)

query = {
"query": {
"range": {"timestamp": {"gte": week_ago.isoformat()}}
},
"aggs": {
"errors_over_time": {
"date_histogram": {
"field": "timestamp",
"calendar_interval": "day"
}
}
},
"size": 0
}
results = es.search(index="logs", body=query)
for bucket in results["aggregations"]["errors_over_time"]["buckets"]:
print(f"Date: {bucket['key_as_string']}, Count: {bucket['doc_count']}")

This query efficiently analyzes time-series data by first narrowing its scope. It uses a range query to select only logs from the past week, which is much faster than scanning the entire index. Once the data is filtered, the query uses a date_histogram aggregation to group the results based on two key parameters:

• The calendar_interval organizes the logs into daily buckets.
• Setting size to 0 is a crucial optimization that tells Elasticsearch you only want the summary counts, not the individual documents.

The final loop then unpacks these buckets to display a clean, day-by-day count.

Get started with Replit

Now, turn these techniques into a real tool. Describe what you want to build to Replit Agent, like "a dashboard showing daily sales from Elasticsearch" or "a script to bulk-index a CSV of user data."

Replit Agent will write the code, test for errors, and help you deploy the application directly from your browser. Start building with Replit.