Elastic Transport for Python

Warnings

• breaking Python 3.8 and 3.9 support was removed in version 9.2.0. Python 3.7 support was dropped in 8.15.0.
  Fix: Upgrade to Python 3.10 or newer.
• breaking The exception hierarchy changed significantly in version 8.x. `TransportError` now only covers transport-level errors (e.g., connection timeouts), while `ApiError` must be used for API-specific errors (e.g., index not found).
  Fix: Update exception handling logic to catch `ApiError` for errors originating from the Elasticsearch API.
• gotcha When using `HttpxAsyncHttpNode`, 404 responses no longer raise exceptions by default; instead, the response with status 404 is returned. This changes the expected error handling behavior for 'not found' scenarios.
  Fix: Explicitly check the `meta.status` (HTTP status code) of the response for 404s if you need to treat them as errors.
• gotcha Ensure compatibility with `httpx` by installing `httpx v0.28.0+` if using the `httpx` backend. Older versions may cause compatibility issues.
  Fix: Upgrade your `httpx` dependency to `v0.28.0` or higher if you use `elastic-transport[httpx]`.
• gotcha The `urllib3` library is the default HTTP client backend. To use `requests`, `aiohttp`, or `httpx`, you must install the respective extra dependencies (e.g., `elastic-transport[requests]`) and explicitly pass the corresponding `NodeConfig` or `node_class` to the `Transport` constructor.
  Fix: Install the desired HTTP client extra and configure the `Transport` accordingly, e.g., `Transport(node_configs=[NodeConfig('...', http_compress=True)], node_class=RequestsHttpNode)`.

Install

• pip install elastic-transport Install stable version
• pip install elastic-transport[requests] Install with Requests HTTP client
• pip install elastic-transport[aiohttp] Install with aiohttp HTTP client
• pip install elastic-transport[httpx] Install with httpx HTTP client (sync and async)

Imports

• Transport

  from elastic_transport import Transport

  The core class for managing nodes and performing HTTP requests.
• RequestsHttpNode

  from elastic_transport import RequestsHttpNode

  To use 'requests' as the underlying HTTP client.
• AiohttpHttpNode

  from elastic_transport import AiohttpHttpNode

  To use 'aiohttp' as the underlying asynchronous HTTP client.
• HttpxHttpNode

  from elastic_transport import HttpxHttpNode

  To use 'httpx' as the underlying synchronous HTTP client.
• HttpxAsyncHttpNode

  from elastic_transport import HttpxAsyncHttpNode

  To use 'httpx' as the underlying asynchronous HTTP client.
• ApiError

  from elastic_transport import ApiError

  For handling API-specific errors, introduced in version 8.x.
• TransportError

  from elastic_transport import TransportError

  For handling transport-level errors (e.g., connection issues).

Quickstart

This quickstart demonstrates how to initialize the `Transport` class, either connecting to a local Elasticsearch instance or an Elastic Cloud deployment using environment variables. It then performs a basic GET request to the cluster root and prints connection details. The `transport.close()` method should be called to close HTTP connections when the transport is no longer needed.

import os
from elastic_transport import Transport, NodeConfig

# Configure nodes - replace with your Elasticsearch host(s)
# For local testing, default to 'http://localhost:9200'
# For cloud, use CLOUD_ID and API_KEY environment variables
cloud_id = os.environ.get('ELASTIC_CLOUD_ID', '')
api_key = os.environ.get('ELASTIC_API_KEY', '')

node_configs = [
    NodeConfig("http://localhost:9200")
]

if cloud_id and api_key:
    # Use Cloud ID and API Key for Elastic Cloud
    transport = Transport(
        node_configs=[], # node_configs are derived from cloud_id
        cloud_id=cloud_id,
        api_key=(api_key,)
    )
else:
    transport = Transport(node_configs=node_configs)

try:
    # Perform a GET request to the root ('/') which often returns cluster info
    # The response is a tuple of (ApiResponseMeta, deserialized_body)
    meta, body = transport.perform_request("GET", "/")
    print("Successfully connected to Elasticsearch!")
    print(f"Cluster Name: {body.get('cluster_name')}")
    print(f"Status Code: {meta.status}")
except Exception as e:
    print(f"Failed to connect or perform request: {e}")
    print("Ensure Elasticsearch is running and accessible at the specified host(s)")
    print("Or check your ELASTIC_CLOUD_ID and ELASTIC_API_KEY environment variables.")

transport.close()