Elasticsearch Adapter Tutorial

This guide demonstrates how to use the ArchiPy Elasticsearch adapter for document indexing, searching, and cluster management with proper error handling and Python 3.14+ type hints.

Installation

uv add "archipy[elasticsearch]"

Tip

The Elasticsearch adapter is an optional extra. Only install it when you need Elasticsearch support to keep your environment lean.

Configuration

Configure Elasticsearch via environment variables or an ElasticsearchConfig object directly.

Environment Variables

Warning

Keep ELASTIC__HTTP_PASSWORD and ELASTIC__API_SECRET out of source control. Prefer API key authentication over basic auth.

Note

The config defaults are REQUEST_TIMEOUT=1.0 seconds and MAX_RETRIES=1. The values below are examples; raise the timeout for slow queries such as large aggregations.

# Basic connection
ELASTIC__HOSTS='["https://localhost:9200"]'

# Basic auth (alternative to API key)
ELASTIC__HTTP_USER_NAME=elastic
ELASTIC__HTTP_PASSWORD=your-password

# API key auth (preferred over basic auth)
ELASTIC__API_KEY=your-api-key-id
ELASTIC__API_SECRET=your-api-key-secret

# TLS/SSL
ELASTIC__CA_CERTS=/path/to/ca.crt
ELASTIC__VERIFY_CERTS=true

# Connection behaviour
ELASTIC__REQUEST_TIMEOUT=5.0
ELASTIC__MAX_RETRIES=3
ELASTIC__RETRY_ON_TIMEOUT=true

Direct Configuration

import logging

from archipy.configs.config_template import ElasticsearchConfig
from archipy.models.errors import ConfigurationError
from pydantic import ValidationError

logger = logging.getLogger(__name__)

try:
    es_config = ElasticsearchConfig(
        HOSTS=["https://localhost:9200"],
        # API key auth (takes precedence over HTTP basic auth)
        API_KEY="my-key-id",
        API_SECRET="my-key-secret",  # type: ignore[arg-type]
        # TLS
        CA_CERTS="/path/to/ca.crt",
        VERIFY_CERTS=True,
        # Timeouts and retries
        REQUEST_TIMEOUT=5.0,
        MAX_RETRIES=3,
        RETRY_ON_TIMEOUT=True,
        RETRY_ON_STATUS=(429, 502, 503, 504),
        # Connection pool
        CONNECTIONS_PER_NODE=10,
        HTTP_COMPRESS=True,
    )
except ValidationError as e:
    logger.error(f"Invalid Elasticsearch configuration: {e}")
    raise ConfigurationError() from e
else:
    logger.info("Elasticsearch configuration created successfully")

Basic Usage

SyncAsync

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

# Uses global config (BaseConfig.global_config().ELASTIC)
try:
    es = ElasticsearchAdapter()
except InternalError as e:
    logger.error(f"Failed to create Elasticsearch adapter: {e}")
    raise
else:
    logger.info("Elasticsearch adapter initialised")

To use a custom config instead of the global one:

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.configs.config_template import ElasticsearchConfig

custom_config = ElasticsearchConfig(HOSTS=["https://es.internal:9200"])
es = ElasticsearchAdapter(elasticsearch_config=custom_config)

import asyncio
import logging

from archipy.adapters.elasticsearch.adapters import AsyncElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)


async def main() -> None:
    try:
        es = AsyncElasticsearchAdapter()
        pong = await es.ping()
        logger.info(f"Cluster reachable: {pong}")
    except InternalError as e:
        logger.error(f"Elasticsearch connection failed: {e}")
        raise


asyncio.run(main())

Index Management

Create an Index

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

index_body = {
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 1,
        "analysis": {
            "analyzer": {
                "custom_analyzer": {
                    "type": "custom",
                    "tokenizer": "standard",
                    "filter": ["lowercase", "stop"],
                }
            }
        },
    },
    "mappings": {
        "properties": {
            "title": {"type": "text", "analyzer": "custom_analyzer"},
            "content": {"type": "text"},
            "author": {"type": "keyword"},
            "published_at": {"type": "date"},
            "view_count": {"type": "integer"},
            "tags": {"type": "keyword"},
        }
    },
}

try:
    if not es.index_exists("articles"):
        response = es.create_index("articles", body=index_body)
        logger.info(f"Index created: {response}")
    else:
        logger.info("Index already exists, skipping creation")
except InternalError as e:
    logger.error(f"Failed to create index: {e}")
    raise

Delete an Index

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

try:
    if es.index_exists("articles"):
        es.delete_index("articles")
        logger.info("Index deleted")
    else:
        logger.info("Index does not exist, nothing to delete")
except InternalError as e:
    logger.error(f"Failed to delete index: {e}")
    raise

Document Operations

Index a Document

import logging
from datetime import datetime, timezone

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

article: dict[str, str | int | list[str]] = {
    "title": "Getting Started with ArchiPy",
    "content": "ArchiPy provides clean-architecture building blocks for Python services.",
    "author": "alice",
    "published_at": datetime.now(tz=timezone.utc).isoformat(),
    "view_count": 0,
    "tags": ["python", "architecture", "clean-code"],
}

try:
    # Index with an explicit document ID
    response = es.index(index="articles", document=article, doc_id="article-001")
    logger.info(f"Document indexed: {response['result']}")

    # Index without specifying an ID — Elasticsearch auto-generates one
    response = es.index(index="articles", document=article)
    generated_id: str = response["_id"]
    logger.info(f"Document auto-indexed with id={generated_id}")
except InternalError as e:
    logger.error(f"Failed to index document: {e}")
    raise

Retrieve a Document

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError, NotFoundError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

try:
    response = es.get(index="articles", doc_id="article-001")
    document = response["_source"]
    logger.info(f"Retrieved document: title={document['title']}")
except KeyError as e:
    logger.warning(f"Document not found: {e}")
    raise NotFoundError() from e
except InternalError as e:
    logger.error(f"Failed to retrieve document: {e}")
    raise

Check Document Existence

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

exists = es.exists(index="articles", doc_id="article-001")
logger.info(f"Document exists: {exists}")

Update a Document

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

# Partial update — only the specified fields are changed
update_fields: dict[str, int] = {"view_count": 42}

try:
    response = es.update(index="articles", doc_id="article-001", doc=update_fields)
    logger.info(f"Document updated: {response['result']}")
except InternalError as e:
    logger.error(f"Failed to update document: {e}")
    raise

Delete a Document

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

try:
    response = es.delete(index="articles", doc_id="article-001")
    logger.info(f"Document deleted: {response['result']}")
except InternalError as e:
    logger.error(f"Failed to delete document: {e}")
    raise

Search

Match Query

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

query: dict[str, object] = {
    "query": {
        "match": {
            "content": "clean architecture"
        }
    }
}

try:
    response = es.search(index="articles", query=query)
    hits = response["hits"]["hits"]
    total = response["hits"]["total"]["value"]
    logger.info(f"Found {total} documents")
    for hit in hits:
        logger.info(f"  [{hit['_score']:.2f}] {hit['_source']['title']}")
except InternalError as e:
    logger.error(f"Search failed: {e}")
    raise

Bool Query with Filters

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

query: dict[str, object] = {
    "query": {
        "bool": {
            "must": [
                {"match": {"content": "python"}}
            ],
            "filter": [
                {"term": {"author": "alice"}},
                {"range": {"view_count": {"gte": 10}}},
            ],
        }
    },
    "sort": [{"published_at": {"order": "desc"}}],
    "size": 10,
    "from": 0,
}

try:
    response = es.search(index="articles", query=query)
    hits = response["hits"]["hits"]
    logger.info(f"Filtered results: {len(hits)} documents")
    for hit in hits:
        src = hit["_source"]
        logger.info(f"  {src['title']} by {src['author']} ({src['view_count']} views)")
except InternalError as e:
    logger.error(f"Filtered search failed: {e}")
    raise

Full-Text Search with Highlighting

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

query: dict[str, object] = {
    "query": {
        "multi_match": {
            "query": "clean architecture python",
            "fields": ["title^2", "content"],  # title boosted 2x
        }
    },
    "highlight": {
        "fields": {
            "title": {},
            "content": {"fragment_size": 150, "number_of_fragments": 3},
        }
    },
}

try:
    response = es.search(index="articles", query=query)
    for hit in response["hits"]["hits"]:
        title = hit["_source"]["title"]
        highlights = hit.get("highlight", {}).get("content", [])
        logger.info(f"Title: {title}")
        for fragment in highlights:
            logger.info(f"  ...{fragment}...")
except InternalError as e:
    logger.error(f"Highlighted search failed: {e}")
    raise

Aggregations

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

query: dict[str, object] = {
    "size": 0,  # No document hits — only aggregations
    "aggs": {
        "authors": {
            "terms": {"field": "author", "size": 10}
        },
        "tags": {
            "terms": {"field": "tags", "size": 20}
        },
        "avg_views": {
            "avg": {"field": "view_count"}
        },
    },
}

try:
    response = es.search(index="articles", query=query)
    aggs = response["aggregations"]

    logger.info("Top authors:")
    for bucket in aggs["authors"]["buckets"]:
        logger.info(f"  {bucket['key']}: {bucket['doc_count']} articles")

    logger.info("Top tags:")
    for bucket in aggs["tags"]["buckets"]:
        logger.info(f"  {bucket['key']}: {bucket['doc_count']} articles")

    logger.info(f"Average view count: {aggs['avg_views']['value']:.1f}")
except InternalError as e:
    logger.error(f"Aggregation query failed: {e}")
    raise

Bulk Operations

Use bulk() to index, update, or delete many documents via elasticsearch.helpers.bulk. The helper accepts generator-friendly action iterables, auto-chunks requests, and retries transient failures (including HTTP 429).

Each action uses helper document format with _index, _id, _source, and optional _op_type (index, create, update, delete). The method returns (success_count, error_count_or_error_list).

Bulk Indexing

import logging
from collections.abc import Iterator
from datetime import datetime, timezone

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from elasticsearch import BulkIndexError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()


def article_actions() -> Iterator[dict[str, object]]:
    """Yield helper-format bulk actions without loading all documents into memory."""
    for i in range(1, 101):
        yield {
            "_index": "articles",
            "_id": f"article-{i:03d}",
            "_source": {
                "title": f"Article {i}",
                "author": "bob",
                "view_count": i * 10,
                "published_at": datetime.now(tz=timezone.utc).isoformat(),
                "tags": ["bulk", "import"],
            },
        }


try:
    success_count, errors = es.bulk(actions=article_actions())
    if isinstance(errors, list) and errors:
        logger.error(f"Bulk index completed with {len(errors)} errors: {errors}")
    else:
        logger.info(f"Bulk index successful: {success_count} documents indexed")
except BulkIndexError as e:
    logger.error(f"Bulk index failed: {e}")
    raise

Bulk Update and Delete

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter
from elasticsearch import BulkIndexError

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

actions: list[dict[str, object]] = [
    {
        "_op_type": "update",
        "_index": "articles",
        "_id": "article-001",
        "doc": {"view_count": 150},
    },
    {
        "_op_type": "update",
        "_index": "articles",
        "_id": "article-002",
        "doc": {"view_count": 89},
    },
    {
        "_op_type": "delete",
        "_index": "articles",
        "_id": "article-999",
    },
]

try:
    success_count, errors = es.bulk(actions=actions, stats_only=True)
    logger.info(f"Bulk update/delete complete: {success_count} successes, {errors} errors")
except BulkIndexError as e:
    logger.error(f"Bulk update/delete failed: {e}")
    raise

Tip

Pass stats_only=True when you only need success and error counts instead of per-document error details.

Scan Operations

Use scan() to iterate over large result sets without manually managing the scroll API. The helper wraps scroll requests and yields hits one at a time.

SyncAsync

import logging

from archipy.adapters.elasticsearch.adapters import ElasticsearchAdapter

logger = logging.getLogger(__name__)

es = ElasticsearchAdapter()

query = {"query": {"match": {"author": "bob"}}}

try:
    hit_count = 0
    for hit in es.scan(query=query, index="articles"):
        hit_count += 1
        logger.debug(f"Scanned document {hit['_id']}: {hit['_source']['title']}")
    logger.info(f"Scan complete: {hit_count} documents")
except Exception as e:
    logger.error(f"Scan failed: {e}")
    raise

import asyncio
import logging

from archipy.adapters.elasticsearch.adapters import AsyncElasticsearchAdapter

logger = logging.getLogger(__name__)


async def scan_articles() -> None:
    es = AsyncElasticsearchAdapter()
    query = {"query": {"match": {"author": "bob"}}}

    hit_count = 0
    async for hit in es.scan(query=query, index="articles"):
        hit_count += 1
        logger.debug(f"Scanned document {hit['_id']}: {hit['_source']['title']}")
    logger.info(f"Async scan complete: {hit_count} documents")


asyncio.run(scan_articles())

Note

Setting preserve_order=True in scan() restores deterministic ordering at the cost of scroll performance.

Async Usage

Async Document Indexing and Search

import asyncio
import logging
from datetime import datetime, timezone

from archipy.adapters.elasticsearch.adapters import AsyncElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)


async def index_and_search() -> None:
    es = AsyncElasticsearchAdapter()

    try:
        # Create index if absent
        if not await es.index_exists("articles"):
            await es.create_index("articles", body={
                "mappings": {
                    "properties": {
                        "title": {"type": "text"},
                        "author": {"type": "keyword"},
                        "published_at": {"type": "date"},
                    }
                }
            })
            logger.info("Index created")

        # Index a document
        await es.index(
            index="articles",
            document={
                "title": "Async Elasticsearch with ArchiPy",
                "author": "carol",
                "published_at": datetime.now(tz=timezone.utc).isoformat(),
            },
            doc_id="async-001",
        )
        logger.info("Document indexed")

        # Search
        response = await es.search(
            index="articles",
            query={"query": {"match": {"author": "carol"}}},
        )
        total = response["hits"]["total"]["value"]
        logger.info(f"Found {total} documents by carol")
    except InternalError as e:
        logger.error(f"Async Elasticsearch operation failed: {e}")
        raise


asyncio.run(index_and_search())

Concurrent Async Operations

import asyncio
import logging
from datetime import datetime, timezone

from archipy.adapters.elasticsearch.adapters import AsyncElasticsearchAdapter
from archipy.models.errors import InternalError

logger = logging.getLogger(__name__)


async def parallel_searches() -> None:
    es = AsyncElasticsearchAdapter()

    queries = [
        {"query": {"match": {"author": "alice"}}},
        {"query": {"match": {"author": "bob"}}},
        {"query": {"range": {"view_count": {"gte": 100}}}},
    ]

    try:
        results = await asyncio.gather(
            *[es.search(index="articles", query=q) for q in queries],
            return_exceptions=True,
        )
    except InternalError as e:
        logger.error(f"Parallel search failed: {e}")
        raise

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.error(f"Query {i} failed: {result}")
        else:
            total = result["hits"]["total"]["value"]
            logger.info(f"Query {i}: {total} documents")


asyncio.run(parallel_searches())

Repository Pattern and Advanced Patterns

The repository pattern, index alias strategy, connection health check and OpenTelemetry setup live on their own page: Elasticsearch Patterns.

Troubleshooting

ConnectionError on startup

• Verify ELASTIC__HOSTS includes the correct scheme (https:// or http://).
• Ensure the Elasticsearch cluster is running: curl -u elastic:<password> https://localhost:9200.
• When VERIFY_CERTS=True, confirm CA_CERTS points to a valid CA bundle.

AuthenticationException (401)

• API key auth takes precedence over HTTP basic auth. If both are set, only the API key is used.
• Rotate the API key if it has been revoked or expired.

BulkIndexError during bulk()

• bulk() delegates to elasticsearch.helpers.bulk, which raises BulkIndexError when any action fails and raise_on_error=True (the default).
• Inspect the exception's errors attribute for per-document failure details.
• Common causes: mapping conflict (a field value does not match its declared type), shard unavailability.
• Pass raise_on_error=False to collect failures in the returned error list instead of raising.

High latency under load

• Increase CONNECTIONS_PER_NODE for concurrent workloads.
• Enable HTTP_COMPRESS=True to reduce payload sizes.
• Use bulk() with generator actions instead of individual index() calls for batch writes.
• Set a realistic REQUEST_TIMEOUT to surface slow queries early.
• Use scan() instead of large search() requests when iterating over many documents.

Sniffing warnings

• ElasticsearchConfig logs a warning when sniffing is enabled with a single non-localhost host — this may bypass load balancers. Disable sniffing (SNIFF_ON_START=False, SNIFF_BEFORE_REQUESTS=False) when connecting through a load balancer or Elastic Cloud.

See Also

• Error Handling — Exception handling patterns with proper chaining
• Configuration Management — Elasticsearch configuration setup
• Elasticsearch Patterns — Repository pattern, index aliases, health check, OpenTelemetry
• Observability — OpenTelemetry (replaces Elastic APM)
• BDD feature files — elastic_adapter.feature, elastic_adapter_apm.feature
• API Reference — Full Elasticsearch adapter API documentation
• Elasticsearch Python Client docs