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 check_liveness() 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. check_liveness() is the engine's own SAFE public health probe -- call this, from application code or anywhere else, not is_alive() (below).

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

engine.active  # True after start(), regardless of reachability
await engine.check_liveness()  # None before anything has borrowed the client,
                                # then True/False once it has
ESAsyncBorder().check_status(engine)  # same answer -- the border's opt-in
                                # tri-state surface is a thin synchronous
                                # bridge over this same method. The plain
                                # ESAsyncBorder().check(engine) stays a
                                # two-valued bool (True unless confirmed
                                # dead), the released Operator contract.

O-lazy-loop-ownership (core/.work/theory/loop-ownership.md, O32): the underlying elasticsearch client's async HTTP node binds its connection pool to whichever event loop makes the FIRST real request, and caches it there for the client's whole life. If your application has a threaded or streaming gateway -- a lifecycle loop separate from the loop that actually serves requests -- a health probe that ran on the wrong loop, before any real request had bound the client, would silently claim it for a loop the request path can never reach. check_liveness() closes that door: it only sends a real ping() once it can prove the calling loop already owns the client; before that (or from a genuinely different loop), it answers None ("no evidence yet") instead of guessing, or a signal fed by real request-path traffic once there is one.

is_alive() is the RAW probe underneath -- it always sends a real ping(), unconditionally, on whatever loop it happens to be awaited from, with NO ownership check of its own. Reach for it directly only when you already know you are on the loop the client is bound to (inside a session() borrow's own task, say); every general-purpose caller wants check_liveness() instead. Calling is_alive() itself 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, including the None verdict before anything has bound the client.

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.