Python Flask ElasticSearch – Front Controller and API documentation


Hi, and welcome to the 3d article devoted to the theme:  “How to work with ElasticSearch, Python and Flask”. Previous article (Part 2: Python, Flask, ElasticSearch and docker environment) is located here. Here we will start to investigate Flask skeleton project. But at first it would be great to refresh at mind our architecture scheme from the Part 1: ElasticSearch Python Flask

Search microservice architecture
Search microservice architecture

 All starts from the Controller. So let’s find it at project structure. It is located the api/resources/hotels folder. Here you will find a file with HotelSearchAPI class.

Python Flask HotelSearchAPI class

Below is code of HotelSearchAPI class

from flask_restful import Resource, reqparse, inputs
from flask_apispec.views import MethodResource
from flask_apispec import marshal_with, doc, use_kwargs
from src.model.response.hotel_search_response_shema import HotelSearchResponseSchema
from src.model.request.hotel_search_request_shema import HotelSearchRequestSchema # noqa
from src.dependencies.hotel_search_criteria.hotel_search_criteria_url_builder import HotelSearchCriteriaUrlBuilder
from src.dependencies.hotel_search_criteria.hotel_search_criteria_director import HotelSearchCriteriaDirector
from src.dependencies.hotel_search.search_service import SearchService
from src.dependencies.hotel_search.query_builder import QueryBuilder
from src.model.response.hotel_search_simple_item_shema import HotelSearchSimpleItemSchema

class HotelSearchAPI(MethodResource, Resource):
        description='Search hotels API.',
             "page": {
                 "description": "paging", 
                 "in": "query", 
                 "type": "int", 
                 "default": 1, 
                 "required": False
             "size": {
                 "description": "size", 
                 "in": "query", 
                 "type": "int", 
                 "default": 10, 
                 "required": False
             "n": {
                 "description": "hotel name", 
                 "in": "query", 
                 "type": "string", 
                 "required": False
             "c": {
                 "description": "city name", 
                 "in": "query", 
                 "type": "string", 
                 "required": False
             "lat": {
                 "description": "lat", 
                 "in": "query", 
                 "type": "float", 
                 "required": False
             "lng": {
                 "description": "lng", 
                 "in": "query", 
                 "type": "float", 
                 "required": False
             "fpn": {
                 "description": "free places", 
                 "in": "query", 
                 "type": "bool", 
                 "required": False
             "age": {
                 "description": "age", 
                 "in": "query", "type": 
                     "int", "default": 5, 
                 "required": False
    @doc(description='Search hotels API.', tags=['search'])
    # @use_kwargs(HotelSearchRequestSchema)
    def get(self):
        parser = self.prepareParser()
        request_data = parser.parse_args()

        builder = HotelSearchCriteriaUrlBuilder(request_data)
        director = HotelSearchCriteriaDirector(builder)
        criteria = director.get_criteria()

        search_service = SearchService(QueryBuilder())
        search_results =

        hotel_items_collection = []

        for hotel in search_results:

        result_response = HotelSearchResponseSchema.create_result_response(

        return result_response if request_data else 0, 200

    def prepareParser():
        parser = reqparse.RequestParser(trim=True, bundle_errors=True)

        parser.add_argument('page', type=int, default="")
        parser.add_argument('size', type=int, default="")
        parser.add_argument('n', type=str, default="")
        parser.add_argument('c', type=str, default="")
        parser.add_argument('lat', type=float, default="")
        parser.add_argument('lng', type=float, default="")
        parser.add_argument('fpn', type=inputs.boolean, default="")
        parser.add_argument('age', type=int, default="")

        return parser

Please, take a look at current controller annotations. That is used by the flask_apispec package which is able to read it properly and create ready documentation for our api that would be avaialble at next url: /swagger-ui

# api/

from flask import Flask
from flask_restful import Api
from import HotelSearchAPI
from dotenv import load_dotenv
from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from flask_apispec.extension import FlaskApiSpec


app = Flask(__name__)
app.secret_key = 'udemyESPython'
api = Api(app)
        title='Udemy Python ElasticSearch',
    'APISPEC_SWAGGER_UI_URL': '/swagger-ui/'

api.add_resource(HotelSearchAPI, '/hotels/search')

docs = FlaskApiSpec(app)

if __name__ == '__main__':

from api import routes # noqa

I will try to explain briefly what we have here. We are saying that our API point is able to process all parameters that are defined by the params section from HotelSearchAPI class. That is all familiar properties we already discussed at initial article “ElasticSearch how to build search system”. I will repeat here our main task scheme. (p.s if it is the first article which you start reading from, maybe you would like to read the previous articles in ascending order – in that case you may choose symfony tag at menu – that will display articles related to current theme sorted by date)

ElasticSearch search
ElasticSearch search

In case response we are using a model HotelSearchResponseSchema (look at HotelSearchAPI annotations) to describe our API. We are saying that our response would be represented as a list of hotel_items_collection items

# src/model/response/

from marshmallow import Schema, fields
from src.model.response.hotel_search_simple_item_shema import HotelSearchSimpleItemSchema

class HotelSearchResponseSchema(Schema):
    results = fields.Nested(HotelSearchSimpleItemSchema, many=True)

    def create_result_response(hotel_items_collection: dict):
        return {"results": hotel_items_collection}

Let’s have a look at HotelSearchSimpleItemSchema class inside

# src/model/response/

from marshmallow import Schema, fields
from elasticsearch_dsl.response import Hit

class HotelSearchSimpleItemSchema(Schema):
    name = fields.Str()
    star = fields.Int()

    def create_item_from_es_response(hotel: Hit):
        return {"name":, "star": hotel.stars}

In human words all that means next – after taking all request parameters , API point returns the list of hotels, where every hotel represented by name and star.

So, what is for all those annotations? Lets open our api doc page.

As you see our GET API point was automatically recognized by the flask_apispec package, moreover if we will expand it – you will see all parameters that can be processed by our API point.

Flask API GET parameters

We also  can see how the response looks and even how our models look inside.

Flask API response

Wow, cool isn’t it?  We have got clear documentation, now any front end developer is able to make integration with our microservice as he knows exactly what parameters we expect and what the final response would be. That is how a good API should look like. So, great, we already have a controller with a well documented API. At the next lecture (Part 4: Python Flask Elasticsearch – builder pattern and DTO search criteria object) we will speak about the builder pattern and DTO search criteria object. If you would like to pass all material more fast, then I propose you to view my on-line course at udemy where you will also find full project skeleton. Below is the link to the course. As the reader of that blog you are also getting possibility to use coupon for the best possible low price. Otherwise, please wait at next articles. Thank you for you attention.

architecture cluster docker elasticsearch flask geo high availability java php programming languages python recommendation systems search systems spring boot symfony