Introduction
When you're building search at scale in Python, the bridge between your application and Elasticsearch matters more than you might think. Elasticsearch with Python: practical tips and trade-offs means understanding not just the client library itself, but how it fits into your architecture, handles concurrency, and manages data lifecycle. After running production search systems across e-commerce platforms and SaaS products, we've learned that the right client choice and patterns can save you from subtle bugs and performance cliffs down the road.
elasticsearch-py: sync and async foundations
The official Python client, elasticsearch-py, is your starting point. It ships with both synchronous and asynchronous transports, and choosing between them shapes everything downstream—error handling, deployment patterns, and integration with your web framework.
Synchronous client is straightforward: blocking calls, familiar patterns, works everywhere. Use it when your application is I/O-bound but not heavily concurrent, or when you're running background jobs.
from elasticsearch import Elasticsearch
es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="products", body={"query": {"match": {"name": "laptop"}}})
for hit in response["hits"]["hits"]:
print(hit["_source"])The async variant pairs naturally with FastAPI, aiohttp, or any async framework. It's not faster per-request—network latency dominates—but it lets one thread handle hundreds of concurrent requests by yielding during I/O waits.
from elasticsearch import AsyncElasticsearch
import asyncio
async def search_products(query: str):
es = AsyncElasticsearch(["http://localhost:9200"])
try:
response = await es.search(
index="products",
body={"query": {"match": {"name": query}}}
)
return response["hits"]["hits"]
finally:
await es.close()
# In FastAPI
@app.get("/search")
async def endpoint(q: str):
results = await search_products(q)
return resultsConnection management matters here. Both sync and async clients maintain connection pools. Reuse a single client instance across your application—don't create one per request. For async, store it in app state or use dependency injection; for sync, a module-level singleton or FastAPI Depends works well.
from fastapi import FastAPI, Depends
from elasticsearch import AsyncElasticsearch
app = FastAPI()
es: AsyncElasticsearch = None
@app.on_event("startup")
async def startup():
global es
es = AsyncElasticsearch(["http://localhost:9200"])
@app.on_event("shutdown")
async def shutdown():
await es.close()
async def get_es():
return es
@app.get("/search")
async def search(q: str, client: AsyncElasticsearch = Depends(get_es)):
return await client.search(index="products", body={"query": {"match": {"name": q}}})helpers.bulk for efficient data loading
Indexing documents one at a time is a performance trap. The bulk API batches writes, slashing round-trips and letting Elasticsearch handle backpressure gracefully. The helpers.bulk function abstracts away the complexity of chunking, retries, and error collection.
Start with a generator that yields document dicts:
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
es = Elasticsearch(["http://localhost:9200"])
def docs_generator(items):
for item in items:
yield {
"_index": "products",
"_id": item["id"],
"_source": {
"name": item["name"],
"price": item["price"],
"category": item["category"]
}
}
items = [
{"id": "1", "name": "Laptop", "price": 999, "category": "electronics"},
{"id": "2", "name": "Monitor", "price": 299, "category": "electronics"},
]
success_count, errors = bulk(es, docs_generator(items), chunk_size=500)
print(f"Indexed {success_count} docs, {len(errors)} errors")For large-scale ingestion from a database, stream results and chunk them:
def docs_from_db():
cursor = db.query("SELECT id, name, price FROM products")
for row in cursor:
yield {
"_index": "products",
"_id": row.id,
"_source": {
"name": row.name,
"price": row.price,
}
}
bulk(es, docs_from_db(), chunk_size=1000)Error handling is critical: bulk returns a list of failed documents. Don't ignore it. For production systems, log errors, filter by error type, and decide whether to retry or skip. Temporarily unavailable shards are recoverable; mapping conflicts are not.
success, errors = bulk(es, docs_generator(items), chunk_size=500, raise_on_error=False)
for error in errors:
if error["index"]["status"] == 429: # Too many requests
# Retry with backoff
pass
elif error["index"]["status"] == 400: # Bad request / mapping error
# Log and skip
print(f"Skipping doc {error['index']['_id']}: {error['index']['error']}")
Integration with FastAPI and Django
FastAPI and Django are the two poles of Python web development. Elasticsearch integration differs because their concurrency models differ.
FastAPI + async Elasticsearch is the natural fit. Your entire stack is async, so use AsyncElasticsearch and let it cooperate with your framework's event loop:
from fastapi import FastAPI
from elasticsearch import AsyncElasticsearch
from pydantic import BaseModel
app = FastAPI()
es = None
@app.on_event("startup")
async def init_es():
global es
es = AsyncElasticsearch(["http://localhost:9200"])
class SearchRequest(BaseModel):
q: str
size: int = 10
@app.post("/search")
async def search(req: SearchRequest):
results = await es.search(
index="products",
body={
"query": {"multi_match": {"query": req.q, "fields": ["name", "description"]}},
"size": req.size
}
)
return {"total": results["hits"]["total"]["value"], "hits": results["hits"]["hits"]}Django + sync Elasticsearch keeps things simpler for traditional request-response cycles. Use sync client in views or management commands. For heavy indexing, run it in a Celery task or Django async view if your Django version supports it.
from django.http import JsonResponse
from elasticsearch import Elasticsearch
from myapp.models import Product
es = Elasticsearch(["http://localhost:9200"])
def search_view(request):
q = request.GET.get("q", "")
results = es.search(
index="products",
body={"query": {"match": {"name": q}}, "size": 20}
)
return JsonResponse({
"total": results["hits"]["total"]["value"],
"results": [hit["_source"] for hit in results["hits"]["hits"]]
})
# Management command for indexing
from django.core.management.base import BaseCommand
from elasticsearch.helpers import bulk
class Command(BaseCommand):
def handle(self, *args, **options):
def docs():
for product in Product.objects.all():
yield {
"_index": "products",
"_id": product.id,
"_source": {
"name": product.name,
"price": product.price,
"category": product.category
}
}
success, errors = bulk(es, docs(), chunk_size=500)
self.stdout.write(f"Indexed {success}, errors: {len(errors)}")Pydantic models for document structure
Raw dicts work, but Pydantic models give you validation, IDE autocomplete, and schema documentation. Define your document structure once and reuse it for indexing and parsing responses.
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class ProductDocument(BaseModel):
id: str
name: str
description: Optional[str] = None
price: float
category: str
tags: list[str] = Field(default_factory=list)
created_at: datetime
in_stock: bool = True
class Config:
schema_extra = {
"example": {
"id": "123",
"name": "Laptop",
"description": "High-performance laptop",
"price": 999.99,
"category": "electronics",
"tags": ["portable", "work"],
"created_at": "2024-01-01T00:00:00",
"in_stock": True
}
}For indexing, convert models to dicts and enrich with metadata:
def docs_from_products(products: list[ProductDocument]):
for product in products:
yield {
"_index": "products",
"_id": product.id,
"_source": product.model_dump(mode="json")
}
# In your indexing pipeline
products = [ProductDocument(...), ProductDocument(...)]
bulk(es, docs_from_products(products), chunk_size=500)For parsing responses, validate hits against your model:
async def search_products(query: str) -> list[ProductDocument]:
response = await es.search(
index="products",
body={"query": {"match": {"name": query}}}
)
results = []
for hit in response["hits"]["hits"]:
try:
doc = ProductDocument(**hit["_source"])
results.append(doc)
except ValueError as e:
# Log schema mismatch but don't crash
logger.warning(f"Invalid document {hit['_id']}: {e}")
return resultsCommon mistakes
- Creating a new client per request. The overhead is substantial and connection pools don't get to do their job. Instantiate once, reuse everywhere.
- Ignoring bulk errors. A failed bulk operation doesn't crash—it returns errors. If you don't check them, invalid or duplicate documents silently don't get indexed.
- Mixing sync and async carelessly. Don't call sync client in an async context or vice versa. If you must, use
asyncio.to_threador a thread pool executor, but that defeats the purpose of async. - Not closing connections. Both sync and async clients should be explicitly closed or used as context managers. Dangling connections exhaust server resources.
- Hardcoding index names. Use configuration and environment variables. You'll want different indices for dev, staging, and production—or the same index with different aliases. Hardcoding makes that impossible.
- Assuming all 200 responses are successes. Elasticsearch returns 200 even if some bulk operations failed. Always parse the response body.
Key takeaways
- Choose sync or async based on your framework and concurrency model, then stick with it throughout your codebase.
- Use
helpers.bulkfor any multi-document operation; it's faster, handles backpressure, and collects errors for you to inspect. - Manage connections explicitly—create one client instance at startup and close it at shutdown. Use dependency injection in FastAPI and module-level singletons in Django.
- Validate documents with Pydantic models to catch schema mismatches early and improve maintainability as your index evolves.
- Always inspect bulk errors; a high success count doesn't mean all documents were indexed correctly.
- For Django, run indexing in management commands or Celery tasks to keep it out of request-response cycles.
Comments
Loading comments…