Skip to main content

Python - Working with the Elasticsearch REST API

Sham Sui Po, Hong Kong

Github Repository

Search - cURL to Python​

curl -XGET "https://localhost:9200/_search?q=Continuous%20Integration&filter_path=took,hits.hits._id,hits.hits._score,hits.hits._source&_source=title&pretty" --insecure -H 'Content-Type: application/json' -u admin

In Python again, I am running into the HTTPS/Certificate issue on localhost and have to add a , verify=False to the request:

requests.exceptions.SSLError: HTTPSConnectionPool(host='localhost', port=9200): Max retries exceeded with url: /_search?q=Continuous%20Integration&filter_path=took,hits.hits._id,hits.hits._score,hits.hits._source&_source=title&pretty (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:997)')))

import requests
import json
from requests.auth import HTTPBasicAuth

username = 'admin'
password = 'admin'

response = requests.get('https://localhost:9200/_search?q=Continuous%20Integration&filter_path=took,hits.hits._id,hits.hits._score,hits.hits._source&_source=title', auth = HTTPBasicAuth(username, password), verify=False)

print(response.content)

This returns the correctly filtered byte response:

b'{"took":5,"hits":{"hits":[{"_id":"docs_ci-cd","_score":3.0692163,"_source":{"title":"What is meant by CI/CD?"}},{"_id":"docs_dev-intro","_score":0.7032547,"_source":{"title":"Application Development and Server Operation"}}]}}'

To extract the JSON string:

result = response.content.decode()
print(type(result))

result_dictionary = json.loads(result)
print(type(result_dictionary))

best_match = result_dictionary['hits']['hits'][0]
print(best_match)

This decoded the bytes into a string and extracted the JSON object from it:

<class 'str'>
<class 'dict'>
{'_id': 'docs_ci-cd', '_score': 3.0692163, '_source': {'title': 'What is meant by CI/CD?'}}

Dynamic Search Terms​

import requests
import json
from requests.auth import HTTPBasicAuth

from elastic_config import *

def ask_es(query):
    url = f'https://{elastic_url}/{elastic_index}/_search?q={query}'
    response = requests.get(url, auth = HTTPBasicAuth(elastic_user, elastic_pass), verify=False)
    result = response.content.decode()
    result_dictionary = json.loads(result)
    
    return result_dictionary

results = ask_es(query='Continuous%20Integration')

for result in results['hits']['hits']:
    print(result['_source']['title'])
    print(result['_source']['abstract'])

with elastic_config.py:

elastic_user = 'admin'
elastic_pass = 'admin'
elastic_url = 'localhost:9200'
elastic_index = 'dev_2022_08_20'

Elasticsearch Flask Client​

from flask import Flask, render_template, request
import urllib.parse
import requests
import json

elastic_user = 'admin'
elastic_pass = 'admin'

def ask_elastic(query, elastic_url, elastic_index):
    url = f'https://{elastic_url}/{elastic_index}/_search?q={query}'
    response = requests.get(url, auth = requests.auth.HTTPBasicAuth(elastic_user, elastic_pass), verify=False)
    result = response.content.decode()
    result_dictionary = json.loads(result)
    
    return result_dictionary

app = Flask(__name__)

@app.route('/', methods=['GET'])
def home_get():
    return render_template('index.html')

@app.route('/', methods=['POST'])
def home_post():
    elastic_url = request.form['elastic_url']
    elastic_index = request.form['elastic_index']
    query = request.form['query']
    safe_request = urllib.parse.quote(query)
    # print(safe_request)
    response = ask_elastic(query=safe_request, elastic_url=elastic_url, elastic_index=elastic_index)
    results = response['hits']['hits']
    return render_template('results.html', query = query, len = len(results), result = results)

app.run(host='0.0.0.0')

Python - Working with the Elasticsearch REST API

index.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Elastic Flask</title>
  </head>
  <body>
    <header>
      <h1>Elasticsearch Client</h1>
    </header>
    <main>
      <p>Enter a search request:</p>
      <form method="POST">
        <div><p>Elasticsearch URL : </p><input placeholder="Elasticsearch URL" value="localhost:9200" name="elastic_url"></input></div>
        <div><p>Elasticsearch Index : </p><input placeholder="Elasticsearch Index" value="dev_2022_08_20" name="elastic_index"></input></div>
        <div><p>Search Query : </p><input placeholder="Search Query" name="query"></input></div>
        <br/>
        <button>Ask Elastic</button>
      </form>
    </main>
  </body>
</html>

Python - Working with the Elasticsearch REST API

results.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Elastic Flask</title>
  </head>
  <body>
    <header>
      <h1>Elasticsearch Client</h1>
    </header>
    <main>
      <p>Enter a search request:</p>
      <form method="POST">
        <div><input value="{{query}}" name="query"></input></div>
        <h3>Search Results</h3>
        <ol>
          {%for i in range(0, len)%}
              <li>{{result[i]['_source']['title']}}</li>
          {%endfor%}
        </ol>
      </form>
    </main>
  </body>
</html>