Hi, and welcome to the 4th article devoted to the theme: “How to work with ElasticSearch, Python and Flask”. Previous article (Part 3: Python Flask ElasticSearch – Front Controller and API documentation) is located here. At that article we will speak about how to transform our requests parameters at the DTO criteria object. That will create an abstract layer between the front world and our backend logic and will allow us to be flexible with coming changes from the front side or in case some additional outside integrations. Here we will use the builder pattern.
We will have a director class. You may treat it as the builder ‘s manager. And HoteSearchCriteria builder that will take url parameters and transform it to a DTO object. Let’s open our project and look at realization. All the classes responsible for transformation logic are located at the dependencies -> hotel_search_criteria folder.
Here you see that we define abstract classes for our directors and builders. While looking at abstract criteria builder you can find get_criteria and create_criteria methods which are synonymous of getResult and and buildPart from builder design scheme presented above. Logic is the same – simply naming is adjusted for our requirements.
import abc
class AbstractCriteriaBuilder(abc.ABC):
@abc.abstractmethod
def get_criteria(self):
pass
@abc.abstractmethod
def create_criteria(self):
passAnd here is our Director class
import abc
class AbstractCriteriaDirector(abc.ABC):
@abc.abstractmethod
def build_criteria(self):
pass
@abc.abstractmethod
def get_criteria(self):
passNow lets investigate realization of abstracts. In case HotelSearchCriteriaDirector – there is nothing special, it is rather simple logic. At real practice it can be more complicated and build final output using additional operations e.g getting some additional data form microservices.
from src.dependencies.hotel_search_criteria.abstract_criteria_builder import AbstractCriteriaBuilder
from src.dependencies.hotel_search_criteria.abstract_criteria_director import AbstractCriteriaDirector
from src.model.criteria.hotel_search_criteria import HotelSearchCriteria
class HotelSearchCriteriaDirector(AbstractCriteriaDirector):
def __init__(self, builder: AbstractCriteriaBuilder):
self.builder = builder
def build_criteria(self):
self.builder.create_criteria()
def get_criteria(self) -> HotelSearchCriteria:
return self.builder.get_criteria()Now lets have a look at our criteria builder realization
from src.dependencies.hotel_search_criteria.abstract_criteria_builder import AbstractCriteriaBuilder
from src.model.criteria.hotel_search_criteria import HotelSearchCriteria
PAGE = 'page'
SIZE = 'size'
HOTEL_NAME = "n"
HOTEL_CITY_NAME_EN = "c"
CITY_CENTER_LAT = "lat"
CITY_CENTER_LNG = "lng"
HOTEL_STARS = "stars"
HOTEL_FREE_PLACES = "fpn"
HOTEL_AGE = "age"
class HotelSearchCriteriaUrlBuilder(AbstractCriteriaBuilder):
def __init__(self, data: dict):
self.data = data
self.criteria = HotelSearchCriteria()
def create_criteria(self):
if PAGE in self.data and isinstance(self.data[PAGE], int):
self.criteria.page = self.data[PAGE]
if SIZE in self.data and isinstance(self.data[SIZE], int) \
and self.data[SIZE] <= HotelSearchCriteria.SIZE_MAX:
self.criteria.size = self.data[SIZE]
if HOTEL_AGE in self.data and isinstance(self.data[HOTEL_AGE], int):
self.criteria.hotel_age = self.data[HOTEL_AGE]
if HOTEL_STARS in self.data and isinstance(self.data[HOTEL_STARS], int):
self.criteria.hotel_stars = self.data[HOTEL_STARS]
if HOTEL_NAME in self.data and isinstance(self.data[HOTEL_NAME], str):
self.criteria.hotel_name = self.data[HOTEL_NAME]
if HOTEL_CITY_NAME_EN in self.data \
and isinstance(self.data[HOTEL_CITY_NAME_EN], str):
self.criteria.city_name = self.data[HOTEL_CITY_NAME_EN]
if HOTEL_FREE_PLACES in self.data \
and isinstance(self.data[HOTEL_FREE_PLACES], bool):
self.criteria.free_places_at_now = self.data[HOTEL_FREE_PLACES]
if (
CITY_CENTER_LAT in self.data
and isinstance(self.data[CITY_CENTER_LAT], float)
and CITY_CENTER_LNG in self.data
and isinstance(self.data[CITY_CENTER_LNG], float)
):
self.criteria.geo_coordinates = {
"lat": self.data[CITY_CENTER_LAT],
"lon": self.data[CITY_CENTER_LNG]
}
def get_criteria(self) -> HotelSearchCriteria:
return self.criteria
At constructor we a recreating our DTO object which is called HotelSearchCriteria. I created it as simple standalone model class. Though it also can be done as ORM entity which can be persisted at database at e.g for logging and further analytics.
class HotelSearchCriteria():
DEFAULT_PAGE = 1
DEFAULT_SIZE = 10
SIZE_MIN = 1
SIZE_MAX = 20
def __init__(self):
self._page = self.DEFAULT_PAGE
self._size = self.DEFAULT_SIZE
self._free_places_at_now = False
self._hotel_name = None
self._city_name = None
self._hotel_age = None
self._hotel_stars = None
self._geo_coordinates = None
@property
def page(self) -> int:
return self._page
@page.setter
def page(self, value: int):
self._page = value
@property
def size(self) -> int:
return self._size
@size.setter
def size(self, value: int):
self._size = value
@property
def free_places_at_now(self) -> bool:
return self._free_places_at_now
@free_places_at_now.setter
def free_places_at_now(self, value: bool):
self._free_places_at_now = value
@property
def hotel_name(self) -> str:
return self._hotel_name
@hotel_name.setter
def hotel_name(self, value: str):
self._hotel_name = value
@property
def city_name(self) -> str:
return self._city_name
@city_name.setter
def city_name(self, value: str):
self._city_name = value
@property
def hotel_age(self) -> int:
return self._hotel_age
@hotel_age.setter
def hotel_age(self, value: int):
self._hotel_age = value
@property
def hotel_stars(self) -> int:
return self._hotel_stars
@hotel_stars.setter
def hotel_stars(self, value: int):
self._hotel_stars = value
@property
def geo_coordinates(self) -> dict:
return self._geo_coordinates
@geo_coordinates.setter
def geo_coordinates(self, value: dict):
self._geo_coordinates = value
Now lets return to our HotelSearchCriteriaUrlBuilder -> create_criteria method. It performs the main work and transforms the url parameters we get at request to DTO object. In that case it is rather simple transformations – small validation and map operations. But in real practice create criteria method can much be more complicated.
Now lets see how put together all that puzzle. Lets return to Front Controller – here I will represent only main method code without annotations (you may find whole code at previous article)
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, 200So, using builder pattern we are creating HotelSearchCriteria DTO object from request. HotelSearchCriteria would represent the interface for our search service, which should know nothing about where search parameters came from and how they were created. Out search service should perform only one responsibility – make a search using DTO parameters.
At the next lecture (Part 5: Python Flask ElasticSearch – model data layer) we will go further with creating our search microservice and speak about dsl python package and Elasticsearch model data layer. 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.
