Rest api added to store and fetch data from elasticsearch.

Kafka sends a message when there is an exception

Author: cloudintro

README.md

@@ -45,7 +45,7 @@ Alternatively we can download the tar.gz package from the Download page and run
 ## 5. Start Elasticsearch server
 Elasticsearch is a distributed, real-time, search analysis platform.
 Elasticsearch can store data in json format, and hence can be used as NoSQL database.
-> elasticsearch-7.11.1/bin/elasticsearch
+> bin/elasticsearch.bat
 
 - index: An index is equivalent to database in relational database
 - mapping: A mapping is equivalent to schema in relational database
@@ -55,7 +55,7 @@ Elasticsearch can store data in json format, and hence can be used as NoSQL data
     > http://localhost:9200
 
     6.2 Index APIs
-        6.2.1 Create an index with name quick_data_index
+        6.2.1 Create an index with name articles
         > PUT http://localhost:9200/articles
 
         6.2.2 Query an index with name articles

quick-data/.gitignore renamed to article-search/.gitignore

@@ -0,0 +1,9 @@
+## article-search service
+
+- This is a demo application to save, and search the article information using elasticsearch.
+- kafka is used to send error message in case of exception as an alarm.
+- article-service is made restful using FastAPI and uvicorn.
+- Read the [README](https://github.com/smallintro/python-elasticsearch-with-kafka/README) file to know how to set up the environment to run and test this application.
+- Run the main.py file to start the service.
+- Access the rest service at [127.0.0.1:8080/docs](http://127.0.0.1:8080/docs)
+

article-search/README.md

@@ -3,7 +3,7 @@ starlette
 pydantic
 fastapi
 uvicorn
-kafka-python
 urllib3
 certifi
+kafka-python
 elasticsearch

quick-data/app/.gitignore renamed to article-search/app/.gitignore

@@ -0,0 +1,49 @@
+from fastapi import FastAPI
+from starlette import status
+import service.app_service as app_service
+from model.data_model import ArticleInfo, AppResponse
+from config.es_config import es_obj
+
+app_v1 = FastAPI()
+
+
+@app_v1.post("/v1/article/save", status_code=status.HTTP_201_CREATED, response_model=AppResponse)
+def save_article_info(article: ArticleInfo):
+    print(f"save_article_info: {article}")
+    msg, data = app_service.save_article_info(article)
+    return AppResponse(status=msg, data=data)
+
+
+@app_v1.get("/v1/article/{article_id}", status_code=status.HTTP_200_OK, response_model=AppResponse)
+def get_article_by_id(article_id: str):
+    print(f"get_article_by_id: {article_id}")
+    msg, data = app_service.get_article_by_id(article_id)
+    return AppResponse(status=msg, data=data)
+
+
+@app_v1.post("/v1/article/find", status_code=status.HTTP_200_OK, response_model=AppResponse)
+def get_article_by_condition(article: ArticleInfo):
+    print(f"get_all_article_info")
+    msg, data = app_service.get_article_by_condition(article)
+    return AppResponse(status=msg, data={'articles': data})
+
+
+@app_v1.get("/v1/article/", status_code=status.HTTP_200_OK, response_model=AppResponse)
+def get_all_articles():
+    print(f"get_all_articles")
+    msg, data = app_service.get_all_articles()
+    return AppResponse(status=msg, data={'articles': data})
+
+
+@app_v1.delete("/v1/article/{article_id}", status_code=status.HTTP_200_OK, response_model=AppResponse)
+def del_article_by_id(article_id: str):
+    print(f"del_article_by_id: {article_id}")
+    msg, data = app_service.del_article_by_id(article_id)
+    return AppResponse(status=msg, data={'result': data})
+
+
+# This gets called once the app is shutting down.
+@app_v1.on_event("shutdown")
+async def app_shutdown():
+    print('closing elasticsearch connection')
+    es_obj.close()

quick-data/app/requirements.txt renamed to article-search/app/requirements.txt

@@ -1,9 +1,8 @@
 from elasticsearch import Elasticsearch
-import json
-import logging
 import requests
 
 _es_obj = None
+default_index = 'articles'
 
 
 def init_elasticsearch():
@@ -13,6 +12,7 @@ def init_elasticsearch():
         print('Elasticsearch Connected')
         res = requests.get('http://localhost:9200')
         print(res.content)
+        create_index()
         return True
     else:
         print('Elasticsearch not connected')
@@ -26,9 +26,8 @@ def es_obj():
     return _es_obj
 
 
-def create_index(index_name):
+def create_index(index_name=default_index):
     result = False
-    # index settings
     settings = {
         "settings": {
             "number_of_shards": 1,
@@ -50,50 +49,11 @@ def create_index(index_name):
             # Ignore 400 means to ignore "Index Already Exist" error.
             es_obj().indices.create(index=index_name, ignore=400, body=settings)
             print(f'Created Index {index_name}')
+        else:
+            es_obj().indices.refresh(index=index_name)
         result = True
     except Exception as ex:
         print('Error while creating index: %s' % str(ex))
     finally:
         return result
 
-
-def add_doc_to_index(index_name, document_data):
-    is_stored = True
-    try:
-        response = es_obj().index(index=index_name, doc_type='_doc', body=json.dumps(document_data))
-        print(response)
-    except Exception as ex:
-        print('Error in indexing data: %s' % str(ex))
-        is_stored = False
-    finally:
-        return is_stored
-
-
-def get_doc_from_index(index_name):
-    response = None
-    # https://www.elastic.co/guide/en/elasticsearch/reference/current/full-text-queries.html
-    search_object = {'query': {'match': {'website': 'smallintro.github.io'}}}
-    search_data = json.dumps(search_object)
-    try:
-        response = es_obj().search(index=index_name, body=search_data)
-        print(response)
-    except Exception as ex:
-        print('Error in indexing data: %s' % str(ex))
-    finally:
-        return response
-
-
-if __name__ == "__main__":
-    logging.basicConfig(level=logging.ERROR)
-    _index_name = 'articles'
-    article_data = {
-        "author": "Sushil",
-        "title": "Small intro to elasticsearch python API",
-        "website": "smallintro.github.io",
-        "publish_date": "2021-11-14",
-        "has_video": True,
-    }
-    if init_elasticsearch:
-        create_index(_index_name)
-        add_doc_to_index(_index_name, article_data)
-        get_doc_from_index(_index_name)

quick-data/app/src/__init__.py renamed to article-search/app/src/__init__.py

@@ -0,0 +1,56 @@
+from kafka import KafkaProducer
+from kafka import KafkaConsumer
+import json
+
+bootstrap_servers = ['localhost:9092']
+default_topic = 'python-kafka-topic'
+group_id = 'kafka-group-id'
+
+_text_producer = None
+_text_consumer = None
+_producer = None
+_consumer = None
+
+
+def init_kafka_producer():
+    global _text_producer, _producer
+    _text_producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
+    _producer = KafkaProducer(bootstrap_servers=bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
+    print(f"init_kafka_producer finished")
+
+
+def init_kafka_consumer(topic_name=default_topic):
+    global _text_consumer, _consumer
+    _text_consumer = KafkaConsumer(topic_name, group_id=group_id, bootstrap_servers=bootstrap_servers)
+    _consumer = KafkaConsumer(topic_name, group_id=group_id, bootstrap_servers=bootstrap_servers,
+                              auto_offset_reset='earliest', enable_auto_commit=True,
+                              value_deserializer=lambda x: json.loads(x.decode('utf-8')))
+    print(f"init_kafka_consumer finished")
+
+
+def producer():
+    global _producer
+    if _producer is None:
+        init_kafka_producer()
+    return _producer
+
+
+def text_producer():
+    global _text_producer
+    if _text_producer is None:
+        init_kafka_producer()
+    return _text_producer
+
+
+def consumer():
+    global _consumer
+    if _consumer is None:
+        init_kafka_consumer()
+    return _consumer
+
+
+def text_consumer():
+    global _text_consumer
+    if _text_consumer is None:
+        init_kafka_consumer()
+    return _text_consumer

article-search/app/src/article_api.py

@@ -0,0 +1,23 @@
+"""
+Created on 14-Nov-2021
+@author: Sushil Prasad
+"""
+import logging
+import uvicorn
+import threading
+from service.kafka_consumer import consume_message
+
+
+def init_app():
+    logging.basicConfig(level=logging.ERROR)
+    try:
+        threading.Thread(target=consume_message).start()
+        print('Hello World!')
+    except Exception as ex:
+        print(str(ex))
+
+
+if __name__ == "__main__":
+    init_app()
+    # uvicorn article_api:app --port 8080 --reload
+    uvicorn.run("article_api:app_v1", host="127.0.0.1", port=8080, log_level="info")

quick-data/app/src/config/__init__.py renamed to article-search/app/src/config/__init__.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+
+
+class ArticleInfo(BaseModel):
+    author: str = None
+    title: str = None
+    website: str = None
+    publish_date: str = None
+    has_video: bool = False
+
+
+class AppResponse(BaseModel):
+    status: str
+    data: dict

quick-data/app/src/service/elasticsearch_service.py renamed to article-search/app/src/config/es_config.py

@@ -0,0 +1,65 @@
+import service.elasticsearch_service as es
+import service.kafka_producer as kfk
+from model.data_model import ArticleInfo
+
+
+def save_article_info(article: ArticleInfo):
+    try:
+        response = es.add_doc_to_index(article)
+        if response.get('result') == 'created':
+            return "success", {response['result']: response['_id']}
+        else:
+            return "failed", {response['result']: None}
+    except Exception as ex:
+        # send a failure message to generate alarm
+        kfk.publish_message(str(ex))
+        return "failed", str(ex)
+
+
+def get_article_by_id(article_id):
+    try:
+        data = {}
+        response = es.get_article_by_id(article_id)
+        for hits in response['hits']['hits']:
+            data[hits['_id']] = hits['_source']
+        return "success", data
+    except Exception as ex:
+        # send a failure message to generate alarm
+        kfk.publish_message(str(ex))
+        return "failed", str(ex)
+
+
+def get_all_articles():
+    try:
+        data = {}
+        response = es.get_all_articles()
+        for hits in response['hits']['hits']:
+            data[hits['_id']] = hits['_source']
+        return "success", data
+    except Exception as ex:
+        # send a failure message to generate alarm
+        kfk.publish_message(str(ex))
+        return "failed", str(ex)
+
+
+def del_article_by_id(article_id):
+    try:
+        response = es.delete_article_by_id(article_id)
+        return "success", response['result']
+    except Exception as ex:
+        # send a failure message to generate alarm
+        kfk.publish_message(str(ex))
+        return "failed", str(ex)
+
+
+def get_article_by_condition(article: ArticleInfo):
+    try:
+        data = {}
+        response = es.get_article_by_author(article.author)
+        for hits in response['hits']['hits']:
+            data[hits['_id']] = hits['_source']
+        return "success", data
+    except Exception as ex:
+        # send a failure message to generate alarm
+        kfk.publish_message(str(ex))
+        return "failed", str(ex)