Python ElasticSearch programming
ElasticSearch connections, indexing, searching, aggregations and TLS from Python services.
Create an outgoing ElasticSearch connection in Dashboard and everything that ElasticSearch offers will be available to your services via the underlying elasticsearch library, as in the examples below.
Creating a connection
In Dashboard, go to Connections -> Outgoing -> ElasticSearch and click Create a new outgoing ElasticSearch connection.
Fill in the form:
- Name - any name of your choice
- Addresses - one full
http(s)://host:portURL per line, e.g. a singlehttps://localhost:9200line or several lines when connecting to a cluster - the scheme of each URL decides whether TLS is used - Username and password - credentials to authenticate with, leave them empty if the server does not require authentication
- Timeout - how many seconds to wait for a response to a single request
The TLS section lets you connect to servers that require encrypted connections - point the CA certificates file to the PEM certificate of the authority that signed the server's certificate. For mutual TLS, add a combined client certificate and private key PEM file too. The validation checkbox can be turned off with test environments whose certificates cannot be verified.
After the connection is created, click Ping to confirm that Zato can reach the server.
Indexing documents
In your services, look up the connection by name with self.es - what you get back is an ElasticSearch client, so indices and documents are accessed the way the client does it.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self):
# Get a client by the connection's name
conn = self.es['My ElasticSearch Connection']
# Index a document under an ID of our choice
result = conn.index(index='orders', id='order-123', document={'order_id': 123, 'status': 'ready'})
self.logger.info('Indexed: %s', result['result'])
Searching
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self):
conn = self.es['My ElasticSearch Connection']
# Get one document by its ID
order = conn.get(index='orders', id='order-123')
self.logger.info('Order: %s', order['_source'])
# Find all documents matching a query
result = conn.search(index='orders', query={'match': {'status': 'ready'}})
for item in result['hits']['hits']:
self.logger.info('Ready: %s', item['_source'])
Updating and deleting
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self):
conn = self.es['My ElasticSearch Connection']
# Update a document
conn.update(index='orders', id='order-123', doc={'status': 'shipped'})
# Delete a document
conn.delete(index='orders', id='order-123')
Aggregations
The full aggregations framework is available too.
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self):
conn = self.es['My ElasticSearch Connection']
# Count orders per status
aggregations = {
'per_status': {
'terms': {'field': 'status.keyword'}
}
}
result = conn.search(index='orders', size=0, aggregations=aggregations)
for bucket in result['aggregations']['per_status']['buckets']:
self.logger.info('%s: %s', bucket['key'], bucket['doc_count'])
Pinging from a service
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class MyService(Service):
def handle(self):
conn = self.es['My ElasticSearch Connection']
# The same request that the Dashboard's Ping link makes
conn.info()
self.logger.info('ElasticSearch is reachable')
Testing with a local server
To test your connections without an existing ElasticSearch installation, one command starts a complete server in Docker:
docker run -d --rm --name zato-elasticsearch -e discovery.type=single-node -e xpack.security.enabled=false -p 9200:9200 docker.elastic.co/elasticsearch/elasticsearch:9.0.0
In Dashboard, create an outgoing ElasticSearch connection with http://localhost:9200 in the addresses field, leave the username and password empty, and click Ping to confirm that Zato can reach the server.
When you are done, the container removes itself on stop: