Greyhorse ElasticSearch library

Greyhorse framework library for Elasticsearch support (async only -- the underlying elasticsearch client is used through its [async] extra).

The primary API is the pieces, not a ready-made module:

An application lists the ones it needs on its own Module, alongside pieces from any other storage library -- no subclassing, no multiple inheritance. See examples/.

How to build

• Install the project

  uv python pin 3.14

  uv sync

  source .venv/bin/activate

• Format and check code

  uv run ruff check --unsafe-fixes --fix

  uv run ruff format

  uv run mypy greyhorse_elasticsearch examples tests

• Run tests

  uv run pytest

Usage

Every snippet below is a runnable program, checked against a live cluster. The longer, commented versions live in examples/ and are executed by the test suite, so they cannot rot silently.

One engine, one client

ESAsyncModule is the ready-made bundle for the single-storage case. The config reaches the engine's constructor through the same args={Type: value} door every greyhorse.strand resource uses -- there is no elasticsearch-specific wiring.

from greyhorse.run import run
from greyhorse.strand import running

from greyhorse_elasticsearch import EngineConf, ESAsyncModule, ESClientCtx


async def main() -> None:
    conf = EngineConf(dsn='http://elastic:elastic@localhost:9200/')

    with running(ESAsyncModule, args={EngineConf: conf}) as module:
        client_ctx = module.get(ESClientCtx).unwrap()
        async with client_ctx as client:
            info = await client.info()
            print(info['cluster_name'])


run(main)

The engine starts when the module starts and stops when it stops; client is an ordinary AsyncElasticsearch borrowed for the length of the async with block. running() stays a plain sync context manager even here -- it starts a module, it is not an I/O operation.

Why there is only one product, and why it is Shared

Redis and SQL siblings publish a second, Mut product (a pipeline, a transaction) whose apply() commits. Elasticsearch has no transaction: every request takes effect the moment the cluster sees it, so there is nothing for an apply()/cancel() pair to mean. The client is therefore Shared -- N consumers may hold it at once -- and the package deliberately does not invent a write window that would only pretend to be one.

Closing is still coordinated: the client is closed once, after the LAST borrow exits. A borrow opened while the engine is closing is refused rather than handed a client about to disappear.

A consumer that knows nothing about greyhorse

The point of the split: the class that talks to Elasticsearch takes a context by TYPE and imports nothing from this package. Only the component says how it is wired.

from typing import ClassVar

from greyhorse.strand import AsyncShared, Component, Use

from greyhorse_elasticsearch import AsyncESEngine, ESAsyncClients, ESAsyncModule, ESClientCtx


class PingApi:
    def __init__(self, client: ESClientCtx) -> None:
        self._client = client

    async def ping(self) -> bool:
        async with self._client as client:
            return bool(await client.ping())


class PingComponent(Component):
    imports: ClassVar = AsyncShared[AsyncESEngine]
    providers: ClassVar = ESAsyncClients
    exports: ClassVar = PingApi


class App(ESAsyncModule):
    name = 'ping-app'
    components: ClassVar = {'ping': Use(PingComponent)}

Subclassing ESAsyncModule fits exactly this shape: one cluster, one consumer, same floor. For two independent clusters -- or Elasticsearch next to a completely different storage -- list the pieces on your own Module instead:

from typing import ClassVar

from greyhorse.strand import Module, Produce, Resource

from greyhorse_elasticsearch import (
    AsyncESEngine,
    ESAsyncBorder,
    ESAsyncClients,
    ESAsyncFragment,
)


class App(Module):
    name = 'search-app'
    fragments: ClassVar = (ESAsyncFragment,)
    resources: ClassVar = (Resource(AsyncESEngine, operators=ESAsyncBorder),)
    produces: ClassVar = (Produce(AsyncESEngine, provider=ESAsyncClients, name='es'),)

examples/03_pieces.py runs that version in full, and explains why a real application composes this way rather than inheriting from several ready-made modules.

Health

.active and is_alive() answer different questions, and confusing them is how a dead cluster reports itself healthy. .active is a start/stop reference count -- it says setup() was called. is_alive() sends a real ping(), bounded by a timeout, cached for a couple of seconds so a health probe cannot stall the tick loop for the duration of the outage it is reporting.

ESAsyncBorder.check() drives the second one, which is what the framework's repair path reads:

engine = ...  # from the module's slot, or ESAsyncEngineFactory().create_engine(...)

engine.active  # True after start(), regardless of reachability
await engine.is_alive()  # False when the cluster cannot be reached
ESAsyncBorder().check(engine)  # same answer, through the framework's road

is_alive() never raises a FAILURE: a refused connection, a TLS failure, a timeout and any driver exception all come back as False. Cancellation is the one deliberate exception -- an external CancelledError propagates rather than being answered as an unhealthy cluster, because a caller who cancelled asked for nothing and must not be handed a verdict. examples/04_liveness.py shows the contrast against an endpoint that was never reachable.

That whole half needs no Module and no wiring at all: ESAsyncEngineFactory().create_engine(...) plus ESAsyncBorder() is enough. examples/04_liveness.py runs exactly that shape, which is the one to copy into your own tests.

Configuration

EngineConf is what the engine is built from:

EngineConf(
    dsn='https://user:password@es.internal:9243/',
    api_key=None,
    request_timeout_seconds=15,
    max_retries=3,
    retry_on_timeout=True,
    verify_certs=True,
    ca_certs=None,
)

verify_certs and ca_certs are passed to the client only for an https DSN -- elastic-transport rejects TLS options on a plain-http node.

ElasticSearchSettings is the environment side. It reads ES_* (and .env), and assembles a DSN from the parts when ES_DSN is not given:

from greyhorse_elasticsearch import ElasticSearchSettings, EngineConf

settings = ElasticSearchSettings()  # ES_HOST, ES_PORT, ES_USER, ...
conf = EngineConf(dsn=settings.dsn)

ES_PASSWORD_FILE points at a Docker/Kubernetes secret file. A missing, unreadable or blank file is an error at config time, not a silent fallback to the inline password. A blank ES_PASSWORD_FILE means "not configured" rather than "read the current directory".

ElasticSearchSettings only assembles a DSN -- pass it into EngineConf yourself for the rest of the engine's tuning knobs.

Credentials. repr() and str() come out redacted -- host and user stay visible, the secret does not -- so a config that reaches a log, an f-string or a traceback does not leak. The same holds for a config that fails validation: ValidationError.errors() and .json(), which is what a JSON logger or an error reporter serializes, carry no credential either. model_dump() deliberately keeps it: that is what builds the client. Dumps are for machines; do not log one.

Tests

The live tests need a running Elasticsearch instance:

docker compose -f tests/docker-compose.yml up -d --wait
export ES_TEST_URI='http://localhost:9200/'
uv run pytest tests -q
docker compose -f tests/docker-compose.yml down -v

They are gated behind the ES_TEST_URI environment variable; without it they skip.