"""
Geo spatial filtering backend.
Elasticsearch supports two types of geo data:
- geo_point fields which support lat/lon pairs
- geo_shape fields, which support points, lines, circles, polygons,
multi-polygons etc.
The queries in this group are:
- geo_shape query: Find document with geo-shapes which either intersect,
are contained by, or do not intersect with the specified geo-shape.
- geo_bounding_box query: Finds documents with geo-points that fall into
the specified rectangle.
+ geo_distance query: Finds document with geo-points within the specified
distance of a central point.
- geo_distance_range query: Like the geo_distance query, but the range
starts at a specified distance from the central point. Note, that this
one is deprecated and this isn't implemented.
+ geo_polygon query: Find documents with geo-points within the specified
polygon.
"""
import logging
from elasticsearch_dsl.query import Q
from rest_framework.filters import BaseFilterBackend
from six import string_types
from ...constants import (
ALL_GEO_SPATIAL_LOOKUP_FILTERS_AND_QUERIES,
LOOKUP_FILTER_GEO_DISTANCE,
LOOKUP_FILTER_GEO_POLYGON,
LOOKUP_FILTER_GEO_BOUNDING_BOX,
SEPARATOR_LOOKUP_COMPLEX_VALUE,
SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE,
)
from ..mixins import FilterBackendMixin
__title__ = 'django_elasticsearch_dsl_drf.filter_backends.filtering.common'
__author__ = 'Artur Barseghyan <artur.barseghyan@gmail.com>'
__copyright__ = '2017-2019 Artur Barseghyan'
__license__ = 'GPL 2.0/LGPL 2.1'
__all__ = ('GeoSpatialFilteringFilterBackend',)
LOGGER = logging.getLogger(__name__)
[docs]class GeoSpatialFilteringFilterBackend(BaseFilterBackend, FilterBackendMixin):
"""Geo-spatial filtering filter backend for Elasticsearch.
Example:
>>> from django_elasticsearch_dsl_drf.constants import (
>>> LOOKUP_FILTER_GEO_DISTANCE,
>>> )
>>> from django_elasticsearch_dsl_drf.filter_backends import (
>>> GeoSpatialFilteringFilterBackend
>>> )
>>> from django_elasticsearch_dsl_drf.viewsets import (
>>> BaseDocumentViewSet,
>>> )
>>>
>>> # Local article document definition
>>> from .documents import ArticleDocument
>>>
>>> # Local article document serializer
>>> from .serializers import ArticleDocumentSerializer
>>>
>>> class ArticleDocumentView(BaseDocumentViewSet):
>>>
>>> document = ArticleDocument
>>> serializer_class = ArticleDocumentSerializer
>>> filter_backends = [GeoSpatialFilteringFilterBackend,]
>>> geo_spatial_filter_fields = {
>>> 'loc': 'location',
>>> 'location': {
>>> 'field': 'location',
>>> 'lookups': [
>>> LOOKUP_FILTER_GEO_DISTANCE,
>>> ],
>>> }
>>> }
"""
[docs] @classmethod
def prepare_filter_fields(cls, view):
"""Prepare filter fields.
:param view:
:type view: rest_framework.viewsets.ReadOnlyModelViewSet
:return: Filtering options.
:rtype: dict
"""
filter_fields = view.geo_spatial_filter_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 'lookups' not in filter_fields[field]:
filter_fields[field]['lookups'] = tuple(
ALL_GEO_SPATIAL_LOOKUP_FILTERS_AND_QUERIES
)
return filter_fields
[docs] @classmethod
def get_geo_distance_params(cls, value, field):
"""Get params for `geo_distance` query.
Example:
/api/articles/?location__geo_distance=2km__43.53__-12.23
:param value:
:param field:
:type value: str
:type field:
:return: Params to be used in `geo_distance` query.
:rtype: dict
"""
__values = cls.split_lookup_complex_value(value, maxsplit=3)
__len_values = len(__values)
if __len_values < 3:
return {}
params = {
'distance': __values[0],
field: {
'lat': __values[1],
'lon': __values[2],
}
}
if __len_values == 4:
params['distance_type'] = __values[3]
else:
params['distance_type'] = 'arc'
return params
[docs] @classmethod
def get_geo_polygon_params(cls, value, field):
"""Get params for `geo_polygon` query.
Example:
/api/articles/?location__geo_polygon=40,-70__30,-80__20,-90
Example:
/api/articles/?location__geo_polygon=40,-70
__30,-80
__20,-90
___name,myname
__validation_method,IGNORE_MALFORMED
Elasticsearch:
{
"query": {
"bool" : {
"must" : {
"match_all" : {}
},
"filter" : {
"geo_polygon" : {
"person.location" : {
"points" : [
{"lat" : 40, "lon" : -70},
{"lat" : 30, "lon" : -80},
{"lat" : 20, "lon" : -90}
]
}
}
}
}
}
}
:param value:
:param field:
:type value: str
:type field:
:return: Params to be used in `geo_distance` query.
:rtype: dict
"""
__values = cls.split_lookup_complex_value(value)
__len_values = len(__values)
if not __len_values:
return {}
__points = []
__options = {}
for __value in __values:
if SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE in __value:
__split_value = __value.split(
SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE
)
if len(__split_value) >= 2:
if __split_value[0] in ('_name', 'validation_method'):
__options.update(
{
__split_value[0]: __split_value[1]
}
)
else:
__points.append(
{
'lat': float(__split_value[0]),
'lon': float(__split_value[1]),
}
)
if __points:
params = {
field: {
'points': __points
}
}
params.update(__options)
return params
return {}
[docs] @classmethod
def get_geo_bounding_box_params(cls, value, field):
"""Get params for `geo_bounding_box` query.
Example:
/api/articles/?location__geo_bounding_box=40.73,-74.1__40.01,-71.12
Example:
/api/articles/?location__geo_polygon=40.73,-74.1
__40.01,-71.12
___name,myname
__validation_method,IGNORE_MALFORMED
__type,indexed
Elasticsearch:
{
"query": {
"bool" : {
"must" : {
"match_all" : {}
},
"filter" : {
"geo_bounding_box" : {
"person.location" : {
"top_left" : {
"lat" : 40.73,
"lon" : -74.1
},
"bottom_right" : {
"lat" : 40.01,
"lon" : -71.12
}
}
}
}
}
}
}
:param value:
:param field:
:type value: str
:type field:
:return: Params to be used in `geo_bounding_box` query.
:rtype: dict
"""
__values = cls.split_lookup_complex_value(value)
__len_values = len(__values)
if not __len_values:
return {}
__top_left_points = {}
__bottom_right_points = {}
__options = {}
# Top left
__lat_lon = __values[0].split(
SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE
)
if len(__lat_lon) >= 2:
__top_left_points.update({
'lat': float(__lat_lon[0]),
'lon': float(__lat_lon[1]),
})
# Bottom right
__lat_lon = __values[1].split(
SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE
)
if len(__lat_lon) >= 2:
__bottom_right_points.update({
'lat': float(__lat_lon[0]),
'lon': float(__lat_lon[1]),
})
# Options
for __value in __values[2:]:
if SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE in __value:
__opt_name_val = __value.split(
SEPARATOR_LOOKUP_COMPLEX_MULTIPLE_VALUE
)
if len(__opt_name_val) >= 2:
if __opt_name_val[0] in ('_name',
'validation_method',
'type'):
__options.update(
{
__opt_name_val[0]: __opt_name_val[1]
}
)
if not __top_left_points or not __bottom_right_points:
return {}
params = {
field: {
'top_left': __top_left_points,
'bottom_right': __bottom_right_points,
}
}
params.update(__options)
return params
[docs] @classmethod
def apply_query_geo_distance(cls, queryset, options, value):
"""Apply `geo_distance` query.
:param queryset: Original queryset.
:param options: Filter options.
:param value: value to filter on.
:type queryset: elasticsearch_dsl.search.Search
:type options: dict
:type value: str
:return: Modified queryset.
:rtype: elasticsearch_dsl.search.Search
"""
return queryset.query(
Q(
'geo_distance',
**cls.get_geo_distance_params(value, options['field'])
)
)
[docs] @classmethod
def apply_query_geo_polygon(cls, queryset, options, value):
"""Apply `geo_polygon` query.
:param queryset: Original queryset.
:param options: Filter options.
:param value: value to filter on.
:type queryset: elasticsearch_dsl.search.Search
:type options: dict
:type value: str
:return: Modified queryset.
:rtype: elasticsearch_dsl.search.Search
"""
return queryset.query(
Q(
'geo_polygon',
**cls.get_geo_polygon_params(value, options['field'])
)
)
[docs] @classmethod
def apply_query_geo_bounding_box(cls, queryset, options, value):
"""Apply `geo_bounding_box` query.
:param queryset: Original queryset.
:param options: Filter options.
:param value: value to filter on.
:type queryset: elasticsearch_dsl.search.Search
:type options: dict
:type value: str
:return: Modified queryset.
:rtype: elasticsearch_dsl.search.Search
"""
return queryset.query(
Q(
'geo_bounding_box',
**cls.get_geo_bounding_box_params(value, options['field'])
)
)
[docs] def get_filter_query_params(self, request, view):
"""Get query params to be filtered on.
: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()
filter_query_params = {}
filter_fields = self.prepare_filter_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 filter_fields:
lookup_param = None
if len(query_param_list) > 1:
lookup_param = query_param_list[1]
valid_lookups = filter_fields[field_name]['lookups']
if lookup_param is None or lookup_param in valid_lookups:
values = [
__value.strip()
for __value
in query_params.getlist(query_param)
if __value.strip() != ''
]
if values:
filter_query_params[query_param] = {
'lookup': lookup_param,
'values': values,
'field': filter_fields[field_name].get(
'field',
field_name
),
'type': view.mapping
}
return filter_query_params
[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
"""
filter_query_params = self.get_filter_query_params(request, view)
for options in filter_query_params.values():
# For all other cases, when we don't have multiple values,
# we follow the normal flow.
for value in options['values']:
# `geo_distance` query lookup
if options['lookup'] == LOOKUP_FILTER_GEO_DISTANCE:
queryset = self.apply_query_geo_distance(
queryset,
options,
value
)
# `geo_polygon` query lookup
elif options['lookup'] == LOOKUP_FILTER_GEO_POLYGON:
queryset = self.apply_query_geo_polygon(
queryset,
options,
value
)
# `geo_bounding_box` query lookup
elif options['lookup'] == LOOKUP_FILTER_GEO_BOUNDING_BOX:
queryset = self.apply_query_geo_bounding_box(
queryset,
options,
value
)
return queryset