"""
Functional suggesters backend.
It's assumed, that fields you're planning to query suggestions for have been
properly indexed using ``fields.CompletionField``.
Example:
>>> from django_elasticsearch_dsl import DocType, Index, fields
>>>
>>> from books.models import Publisher
>>>
>>> # Name of the Elasticsearch index
>>> PUBLISHER_INDEX = Index(PUBLISHER_INDEX_NAME)
>>> # See Elasticsearch Indices API reference for available settings
>>> PUBLISHER_INDEX.settings(
>>> number_of_shards=1,
>>> number_of_replicas=1
>>> )
>>>
>>> @PUBLISHER_INDEX.doc_type
>>> class PublisherDocument(DocType):
>>> "Publisher Elasticsearch document."
>>>
>>> id = fields.IntegerField(attr='id')
>>>
>>> name = fields.StringField(
>>> fields={
>>> 'raw': fields.StringField(analyzer='keyword'),
>>> 'suggest': fields.CompletionField(),
>>> }
>>> )
>>>
>>> info = fields.StringField()
>>>
>>> address = fields.StringField(
>>> fields={
>>> 'raw': fields.StringField(analyzer='keyword')
>>> }
>>> )
>>>
>>> city = fields.StringField(
>>> fields={
>>> 'raw': fields.StringField(analyzer='keyword'),
>>> 'suggest': fields.CompletionField(),
>>> }
>>> )
>>>
>>> state_province = fields.StringField(
>>> fields={
>>> 'raw': fields.StringField(analyzer='keyword'),
>>> 'suggest': fields.CompletionField(),
>>> }
>>> )
>>>
>>> country = fields.StringField(
>>> fields={
>>> 'raw': fields.StringField(analyzer='keyword'),
>>> 'suggest': fields.CompletionField(),
>>> }
>>> )
>>>
>>> website = fields.StringField()
>>>
>>> class Meta(object):
>>> "Meta options."
>>>
>>> model = Publisher # The model associate with this DocType
"""
from elasticsearch_dsl.search import AggsProxy
from django_elasticsearch_dsl_drf.constants import (
FUNCTIONAL_SUGGESTER_TERM_MATCH,
FUNCTIONAL_SUGGESTER_PHRASE_MATCH,
FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX,
FUNCTIONAL_SUGGESTER_COMPLETION_MATCH,
ALL_FUNCTIONAL_SUGGESTERS,
)
from django_elasticsearch_dsl_drf.utils import EmptySearch
from rest_framework.exceptions import ValidationError
from rest_framework.filters import BaseFilterBackend
from six import string_types
from ..mixins import FilterBackendMixin
__title__ = 'django_elasticsearch_dsl_drf.filter_backends.suggester.' \
'functional'
__author__ = 'Artur Barseghyan <artur.barseghyan@gmail.com>'
__copyright__ = '2017-2018 Artur Barseghyan'
__license__ = 'GPL 2.0/LGPL 2.1'
__all__ = ('FunctionalSuggesterFilterBackend',)
[docs]class FunctionalSuggesterFilterBackend(BaseFilterBackend, FilterBackendMixin):
"""Suggester filter backend for Elasticsearch.
Suggestion functionality is exclusive. Once you have queried the
``FunctionalSuggesterFilterBackend``, the latter will transform your
current search query into another search query (altered).
Therefore, always add it as the very last filter backend.
Example:
>>> from django_elasticsearch_dsl_drf.constants import (
>>> FUNCTIONAL_SUGGESTER_COMPLETION_MATCH,
>>> FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX,
>>> FUNCTIONAL_SUGGESTER_PHRASE_MATCH,
>>> FUNCTIONAL_SUGGESTER_PHRASE_MATCH,
>>> FUNCTIONAL_SUGGESTER_TERM_MATCH,
>>> )
>>> from django_elasticsearch_dsl_drf.filter_backends import (
>>> FunctionalSuggesterFilterBackend
>>> )
>>> from django_elasticsearch_dsl_drf.viewsets import DocumentViewSet
>>>
>>> # Local PublisherDocument definition
>>> from .documents import PublisherDocument
>>>
>>> # Local PublisherDocument serializer
>>> from .serializers import PublisherDocumentSerializer
>>>
>>> class PublisherDocumentView(DocumentViewSet):
>>>
>>> document = PublisherDocument
>>> serializer_class = PublisherDocumentSerializer
>>> filter_backends = [
>>> # ...
>>> FunctionalSuggesterFilterBackend,
>>> ]
>>> # Suggester fields
>>> suggester_fields = {
>>> 'name_suggest': {
>>> 'field': 'name.suggest',
>>> 'suggesters': [
>>> FUNCTIONAL_SUGGESTER_COMPLETION_MATCH,
>>> FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX,
>>> ],
>>> },
>>> 'city_suggest': {
>>> 'field': 'city.suggest',
>>> 'suggesters': [
>>> FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX,
>>> ],
>>> },
>>> 'state_province_suggest': {
>>> 'field': 'state_province.suggest',
>>> 'suggesters': [
>>> FUNCTIONAL_SUGGESTER_COMPLETION_MATCH,
>>> ],
>>> },
>>> 'country_suggest': {
>>> 'field': 'country.suggest',
>>> 'suggesters': [
>>> FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX,
>>> ],
>>> },
>>> }
"""
[docs] @classmethod
def prepare_suggester_fields(cls, view):
"""Prepare filter fields.
:param view:
:type view: rest_framework.viewsets.ReadOnlyModelViewSet
:return: Filtering options.
:rtype: dict
"""
filter_fields = view.functional_suggester_fields
for field, options in filter_fields.items():
if options is None or isinstance(options, string_types):
filter_fields[field] = {
'field': options or field
}
elif 'field' not in filter_fields[field]:
filter_fields[field]['field'] = field
if 'suggesters' not in filter_fields[field]:
filter_fields[field]['suggesters'] = tuple(
ALL_FUNCTIONAL_SUGGESTERS
)
return filter_fields
# @classmethod
# def apply_suggester_term(cls, suggester_name, queryset, options, value):
# """Apply `term` suggester.
#
# :param suggester_name:
# :param queryset: Original queryset.
# :param options: Filter options.
# :param value: value to filter on.
# :type suggester_name: str
# :type queryset: elasticsearch_dsl.search.Search
# :type options: dict
# :type value: str
# :return: Modified queryset.
# :rtype: elasticsearch_dsl.search.Search
# """
# return queryset.suggest(
# suggester_name,
# value,
# term={'field': options['field']}
# )
#
# @classmethod
# def apply_suggester_phrase(cls,
# suggester_name,
# queryset,
# options,
# value):
# """Apply `phrase` suggester.
#
# :param suggester_name:
# :param queryset: Original queryset.
# :param options: Filter options.
# :param value: value to filter on.
# :type suggester_name: str
# :type queryset: elasticsearch_dsl.search.Search
# :type options: dict
# :type value: str
# :return: Modified queryset.
# :rtype: elasticsearch_dsl.search.Search
# """
# return queryset.suggest(
# suggester_name,
# value,
# phrase={'field': options['field']}
# )
[docs] @classmethod
def apply_suggester_completion_prefix(cls,
suggester_name,
queryset,
options,
value):
"""Apply `completion` suggester prefix.
This is effective when used with Keyword fields.
:param suggester_name:
:param queryset: Original queryset.
:param options: Filter options.
:param value: value to filter on.
:type suggester_name: str
:type queryset: elasticsearch_dsl.search.Search
:type options: dict
:type value: str
:return: Modified queryset.
:rtype: elasticsearch_dsl.search.Search
"""
queryset = queryset.query(
'prefix',
**{options['field']: value}
)
return queryset
[docs] @classmethod
def apply_suggester_completion_match(cls,
suggester_name,
queryset,
options,
value):
"""Apply `completion` suggester match.
This is effective when used with Ngram fields.
:param suggester_name:
:param queryset: Original queryset.
:param options: Filter options.
:param value: value to filter on.
:type suggester_name: str
:type queryset: elasticsearch_dsl.search.Search
:type options: dict
:type value: str
:return: Modified queryset.
:rtype: elasticsearch_dsl.search.Search
"""
queryset = queryset.query(
'match',
**{options['field']: value}
)
return queryset
[docs] def get_suggester_query_params(self, request, view):
"""Get query params to be for suggestions.
:param request: Django REST framework request.
:param view: View.
:type request: rest_framework.request.Request
:type view: rest_framework.viewsets.ReadOnlyModelViewSet
:return: Request query params to filter on.
:rtype: dict
"""
query_params = request.query_params.copy()
suggester_query_params = {}
suggester_fields = self.prepare_suggester_fields(view)
for query_param in query_params:
query_param_list = self.split_lookup_filter(
query_param,
maxsplit=1
)
field_name = query_param_list[0]
if field_name in suggester_fields:
suggester_param = None
if len(query_param_list) > 1:
suggester_param = query_param_list[1]
valid_suggesters = suggester_fields[field_name]['suggesters']
# If we have default suggester given use it as a default and
# do not require further suffix specification.
default_suggester = None
if 'default_suggester' in suggester_fields[field_name]:
default_suggester = \
suggester_fields[field_name]['default_suggester']
if suggester_param is None \
or suggester_param in valid_suggesters:
# If we have default suggester given use it as a default
# and do not require further suffix specification.
if suggester_param is None \
and default_suggester is not None:
suggester_param = str(default_suggester)
values = [
__value.strip()
for __value
in query_params.getlist(query_param)
if __value.strip() != ''
]
# If specific field given, use that. Otherwise,
# fall back to the top level field name.
if 'serializer_field' in suggester_fields[field_name]:
serializer_field = \
suggester_fields[field_name]['serializer_field']
else:
serializer_field = suggester_fields[field_name].get(
'field',
field_name
)
serializer_field = self.extract_field_name(
serializer_field
)
if values:
suggester_query_params[query_param] = {
'suggester': suggester_param,
'values': values,
'field': suggester_fields[field_name].get(
'field',
field_name
),
'type': view.mapping,
'serializer_field': serializer_field,
}
return suggester_query_params
[docs] def clean_queryset(self, queryset):
"""Clean the queryset.
- Remove aggregations.
- Remove highlight.
- Remove sorting options.
:param queryset:
:return:
"""
queryset.aggs = AggsProxy('')
queryset._highlight = {}
queryset._sort = ['_score']
queryset._functional_suggest = True
return queryset
[docs] def serialize_queryset(self,
queryset,
suggester_name,
value,
serializer_field):
"""Serialize queryset.
This shall be done here, since we don't want to delegate it to
pagination.
:param queryset:
:param suggester_name:
:param value:
:param serializer_field:
:return:
"""
result = queryset.execute().to_dict()
hits = []
for hit in result['hits']['hits']:
hit.update({'text': hit['_source'].get(serializer_field)})
hits.append(hit)
data = {
suggester_name: [{
'text': value,
'options': hits,
'length': len(value),
'offset': 0,
}],
'_shards': result['_shards'],
}
return data
[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: elasticsearch_dsl.search.Search
:type view: rest_framework.viewsets.ReadOnlyModelViewSet
:return: Updated queryset.
:rtype: elasticsearch_dsl.search.Search
"""
# The ``SuggesterFilterBackend`` filter backend shall be used in
# the ``suggest`` custom view action/route only. Usages outside of the
# are ``suggest`` action/route are restricted.
if view.action != 'functional_suggest':
return queryset
# Clean the queryset.
queryset = self.clean_queryset(queryset)
suggester_query_params = self.get_suggester_query_params(request, view)
applied = False # Indicates whether filter has been applied
picked_suggester_name = None
picked_value = None
picked_serializer_field = None
for suggester_name, options in suggester_query_params.items():
# We don't have multiple values here.
for value in options['values']:
# `completion` suggester
if options['suggester'] == \
FUNCTIONAL_SUGGESTER_COMPLETION_PREFIX:
queryset = self.apply_suggester_completion_prefix(
suggester_name,
queryset,
options,
value
)
applied = True
picked_suggester_name = suggester_name
picked_value = value
picked_serializer_field = options['serializer_field']
elif options['suggester'] == \
FUNCTIONAL_SUGGESTER_COMPLETION_MATCH:
queryset = self.apply_suggester_completion_match(
suggester_name,
queryset,
options,
value
)
applied = True
picked_suggester_name = suggester_name
picked_value = value
picked_serializer_field = options['serializer_field']
# # `term` suggester
# elif options['suggester'] == SUGGESTER_TERM:
# queryset = self.apply_suggester_term(suggester_name,
# queryset,
# options,
# value)
#
# # `phrase` suggester
# elif options['suggester'] == SUGGESTER_PHRASE:
# queryset = self.apply_suggester_phrase(suggester_name,
# queryset,
# options,
# value)
# If no filters have been applied, return empty queryset. This
# has no affect on other backends, since this only applies to
# view.action == 'functional_suggest' case.
if not applied:
raise ValidationError(detail=None)
# empty_queryset = EmptySearch()
# empty_queryset._functional_suggest = True
# return empty_queryset
return self.serialize_queryset(
queryset,
picked_suggester_name,
picked_value,
picked_serializer_field
)