Python client

The Lucenia Python client (lucenia-py) provides Python wrappers around the Lucenia REST API so you can interact with your cluster using native Python data structures instead of building raw HTTP requests. The client also includes higher-level helpers for bulk indexing and asynchronous I/O.

Compatibility

The lucenia-py client requires Python 3.8 or later. The PyPI package version tracks the Lucenia server version: client 0.11.x is intended for use with Lucenia 0.11.x clusters.

Installation

Install the client from PyPI using pip:

pip install lucenia-py

For production projects, pin to a specific minor version in your requirements.txt or pyproject.toml:

lucenia-py>=0.11.0,<0.12.0

Then import the client:

from lucenia import Lucenia

Creating a client

Connecting to a secured cluster

The default Lucenia configuration runs with TLS and basic authentication enabled. The snippet below creates a client suitable for local development against a cluster using self-signed certificates. Configure a real CA bundle and enable certificate verification in production.

from lucenia import Lucenia

client = Lucenia(
    hosts=[{"host": "localhost", "port": 9200}],
    http_auth=("admin", "MyStrongPassword123!"),
    use_ssl=True,
    verify_certs=False,
    ssl_show_warn=False,
)

info = client.info()
print(f"Connected to {info['version']['distribution']} {info['version']['number']}")

warning

verify_certs=False disables TLS verification and should only be used for local development. In production, supply a ca_certs path pointing to your cluster's CA certificate and set verify_certs=True.

To enable certificate verification, pass a ca_certs path:

client = Lucenia(
    hosts=[{"host": "localhost", "port": 9200}],
    http_auth=("admin", "MyStrongPassword123!"),
    use_ssl=True,
    verify_certs=True,
    ca_certs="/path/to/root-ca.pem",
)

Connecting to an unsecured cluster

For development clusters running without TLS or authentication:

client = Lucenia(hosts=[{"host": "localhost", "port": 9200}])

Working with documents

Creating an index

Create an index with custom settings and mappings:

index_name = "my-index"

index_body = {
    "settings": {
        "index": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
        }
    },
    "mappings": {
        "properties": {
            "title": {"type": "text"},
            "text":  {"type": "text"},
            "year":  {"type": "integer"},
        }
    },
}

response = client.indices.create(index_name, body=index_body)
print(f"Index created: {response['acknowledged']}")

Indexing a single document

document = {
    "title": "Introduction to Lucenia",
    "text":  "Lucenia is a high-performance search engine.",
    "year":  2025,
}

response = client.index(
    index=index_name,
    id="1",
    body=document,
    refresh=True,
)

print(response["result"])

Bulk indexing

For higher throughput, use the helpers.bulk helper to index many documents in a single request. The helper handles serialization to the bulk newline-delimited JSON format and returns success and error counts:

from lucenia import helpers

actions = [
    {
        "_index": index_name,
        "_id": "2",
        "_source": {"title": "Getting Started", "text": "Use the lucenia-py client.", "year": 2025},
    },
    {
        "_index": index_name,
        "_id": "3",
        "_source": {"title": "Search Features", "text": "Full-text search and more.", "year": 2025},
    },
    {
        "_index": index_name,
        "_id": "4",
        "_source": {"title": "Bulk Operations", "text": "Index many docs at once.", "year": 2025},
    },
]

success_count, errors = helpers.bulk(client, actions)
print(f"Bulk indexed {success_count} documents (errors: {errors})")

# Make the new documents searchable immediately.
client.indices.refresh(index=index_name)

For very large datasets, helpers.parallel_bulk streams batches in parallel and is suitable for indexing millions of documents from a generator.

Searching

Run a match_all query to return every document:

response = client.search(index=index_name, body={"query": {"match_all": {}}})

print(f"Found {response['hits']['total']['value']} documents:")
for hit in response["hits"]["hits"]:
    print(f"  [{hit['_id']}] {hit['_source']['title']}")

Run a match query against a specific field:

query = {
    "query": {
        "match": {
            "title": "Lucenia",
        }
    }
}

response = client.search(index=index_name, body=query)

for hit in response["hits"]["hits"]:
    print(f"[{hit['_id']}] score={hit['_score']:.4f} {hit['_source']['title']}")

Run a multi_match query that boosts matches in the title field:

query = {
    "query": {
        "multi_match": {
            "query":  "search",
            "fields": ["title^2", "text"],
        }
    }
}

response = client.search(index=index_name, body=query)

For pagination, sorting, point-in-time, and scroll search, see the Query DSL documentation.

Deleting a document and the index

client.delete(index=index_name, id="1")

client.indices.delete(index=index_name)

Asynchronous client

For high-concurrency applications, lucenia-py ships an async client that uses aiohttp under the hood. Import AsyncLucenia instead of Lucenia and await each call:

import asyncio
from lucenia import AsyncLucenia

async def main():
    client = AsyncLucenia(
        hosts=[{"host": "localhost", "port": 9200}],
        http_auth=("admin", "MyStrongPassword123!"),
        use_ssl=True,
        verify_certs=False,
    )
    info = await client.info()
    print(info["version"]["number"])
    await client.close()

asyncio.run(main())

Next steps

The Lucenia Python client supports a wider set of features beyond the basics covered here, including:

• Asynchronous I/O with AsyncLucenia
• Authentication with AWS IAM (SigV4)
• Configurable TLS / SSL contexts
• Bulk indexing helpers and parallel_bulk
• High-level DSL for building queries
• Search pagination, point-in-time, and scroll
• Index templates
• k-NN plugin integration