Python Flask ElasticSearch — Front Controller and API documentation — Sergii Demianchuk Blog

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):
@doc(
description='Search hotels API.',
tags=['search'],
params={
"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)
@marshal_with(HotelSearchResponseSchema)
def get(self):
parser = self.prepareParser()
request_data = parser.parse_args()

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

search_service = SearchService(QueryBuilder())
search_results = search_service.search(criteria)

hotel_items_collection = []

for hotel in search_results:
hotel_items_collection.append(
HotelSearchSimpleItemSchema.create_item_from_es_response(hotel)
)

result_response = HotelSearchResponseSchema.create_result_response(
hotel_items_collection
)

return result_response if request_data else 0, 200

@staticmethod
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
# api/__init__.py from flask from flask import Flask
from flask_restful import Api
from api.resources.hotels.hotel_search_api import HotelSearchAPI
from dotenv import load_dotenv
from apispec import APISpec
from apispec.ext.marshmallow import MarshmallowPlugin
from flask_apispec.extension import FlaskApiSpec

load_dotenv()

app = Flask(__name__)
app.secret_key = 'udemyESPython'
api = Api(app)
app.config.update({
'APISPEC_SPEC': APISpec(
title='Udemy Python ElasticSearch',
version='v1',
plugins=[MarshmallowPlugin()],
openapi_version='2.0.0'
),
'APISPEC_SWAGGER_UI_URL': '/swagger-ui/'
})

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

docs = FlaskApiSpec(app)
docs.register(HotelSearchAPI)

if __name__ == '__main__':
app.run(debug=True)

from api import routes # noqa
# src/model/response/hotel_search_response_shema.py 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)

@staticmethod
def create_result_response(hotel_items_collection: dict):
return {"results": hotel_items_collection}
# src/model/response/hotel_search_simple_item_shema.py from marshmallow import Schema, fields
from elasticsearch_dsl.response import Hit


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

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

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Demianchuk Sergii

Demianchuk Sergii

15 year’s experience at IT. I am specialized at architecture for complex systems, search & recommendations systems, ML, devops, security, big data sets analysis