Source code for

"""Base search backend."""

from django.core.exceptions import ImproperlyConfigured

from django_elasticsearch_dsl import fields
from rest_framework.filters import BaseFilterBackend
from rest_framework.settings import api_settings

from ..mixins import FilterBackendMixin
from ...compat import coreapi, coreschema

__title__ = ''
__author__ = 'Artur Barseghyan <>'
__copyright__ = '2017-2020 Artur Barseghyan'
__license__ = 'GPL 2.0/LGPL 2.1'
__all__ = (

[docs]class BaseSearchFilterBackend(BaseFilterBackend, FilterBackendMixin): """Base search filter backend.""" query_backends = [] matching = DEFAULT_MATCHING_OPTION search_param = api_settings.SEARCH_PARAM
[docs] def get_search_query_params(self, request): """Get search query params. :param request: Django REST framework request. :type request: rest_framework.request.Request :return: List of search query params. :rtype: list """ query_params = request.query_params.copy() return query_params.getlist(self.search_param, [])
[docs] def get_query_backends(self, request, view): """Get query backends. :return: """ raise NotImplementedError( "You should define `get_query_backends` method in your {} class" "".format(self.__class__.__name__) )
def _get_query_backends(self, request, view): """Get query backends internal. :param request: :param view: :return: """ try: return self.get_query_backends(request, view) except NotImplementedError as err: pass if not self.query_backends: raise NotImplementedError( "Your search backend shall either implement " "`get_query_backends` method or define `query_backends`" "property." ) return self.query_backends[:]
[docs] def filter_queryset(self, request, queryset, view): """Filter the queryset. :param request: Django REST framework request. :param queryset: Base queryset. :param view: View. :type request: rest_framework.request.Request :type queryset: :type view: rest_framework.viewsets.ReadOnlyModelViewSet :return: Updated queryset. :rtype: """ if self.matching not in MATCHING_OPTIONS: raise ImproperlyConfigured( "Your `matching` value does not match the allowed matching" "options: {}".format(', '.join(MATCHING_OPTIONS)) ) __query_backends = self._get_query_backends(request, view) if len(__query_backends) > 1: __queries = [] for query_backend in __query_backends: __queries.extend( query_backend.construct_search( request=request, view=view, search_backend=self ) ) if __queries: queryset = queryset.query( 'bool', **{self.matching: __queries} ) elif len(__query_backends) == 1: __query = __query_backends[0].construct_search( request=request, view=view, search_backend=self ) queryset = queryset.query('bool', **{self.matching: __query}) else: raise ImproperlyConfigured( "Search filter backend shall have at least one query_backend" "specified either in `query_backends` property or " "`get_query_backends` method. Make appropriate changes to" "your {} class".format(self.__class__.__name__) ) return queryset
[docs] def get_coreschema_field(self, field): if isinstance(field, fields.IntegerField): field_cls = coreschema.Number else: field_cls = coreschema.String return field_cls()
[docs] def get_schema_fields(self, view): assert coreapi is not None, 'coreapi must be installed to ' \ 'use `get_schema_fields()`' assert coreschema is not None, 'coreschema must be installed to ' \ 'use `get_schema_fields()`' _search_fields = getattr(view, 'search_fields', None) if isinstance(_search_fields, dict): search_fields = list(_search_fields.keys()) else: search_fields = _search_fields return [] if not search_fields else [ coreapi.Field( name=self.search_param, required=False, location='query', schema=coreschema.String( description='Search in ' '{}.'.format(', '.join(search_fields)) ) ) ]