Python Flask Elasticsearch – builder pattern and DTO search criteria object

P

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.

Builder design pattern
Builder design 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.

HotelSearchCriteria

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):
        pass

And here is our Director class

import abc


class AbstractCriteriaDirector(abc.ABC):
    @abc.abstractmethod
    def build_criteria(self):
        pass

    @abc.abstractmethod
    def get_criteria(self):
        pass

Now 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, 200

So, 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.


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