import django_filters
from django.conf import settings
from elasticsearch_dsl import Q
from rest_framework.filters import BaseFilterBackend
from apps.posts.documents import PostDocument
from apps.posts.models import Post, Tag
from apps.users.models import Category
[docs]
class FollowingPostsFilter(django_filters.BooleanFilter):
"""
A filter for selecting posts whose authors are in the list to which the user is subscribed.
"""
[docs]
def filter(self, qs, value):
if value is not True:
return qs
user = self.parent.request.user
if user.is_authenticated:
following_users = user.following.all()
return qs.filter(author__in=following_users).order_by("-created_at")
return qs.none()
[docs]
class PostFilter(django_filters.FilterSet):
tags = django_filters.ModelMultipleChoiceFilter(
queryset=Tag.objects.all(), to_field_name="name", conjoined=True, field_name="tags__name", label="List of tags"
)
following_posts = FollowingPostsFilter(label="Only posts whose authors are followed by the user")
categories = django_filters.ModelMultipleChoiceFilter(
queryset=Category.objects.all(), to_field_name="pk", field_name="categories__pk", label="List of categories"
)
class Meta:
model = Post
fields = ["tags", "following_posts", "categories"]
[docs]
class ElasticsearchFilter(BaseFilterBackend):
"""
Filter for performing search using Elasticsearch.
Executes search on multiple fields with different weights.
"""
[docs]
def construct_query(self, search_param):
"""
Constructs a combined Elasticsearch query from various parts.
:param search_param: The search parameter entered by the user.
:return: A combined Elasticsearch query.
"""
# Queries for each parameter
category_and_tag_and_title_query = Q(
"bool",
must=[
Q("match", title={"query": search_param, "fuzziness": "AUTO"}),
Q("match", categories__name={"query": search_param, "fuzziness": "AUTO"}),
Q("match", tags__name={"query": search_param, "fuzziness": "AUTO"}),
],
)
title_query = Q("match", title={"query": search_param, "fuzziness": "AUTO", "boost": 4})
category_and_title_query = Q(
"bool",
must=[
Q("match", title={"query": search_param, "fuzziness": "AUTO"}),
Q("match", categories__name={"query": search_param, "fuzziness": "AUTO", "boost": 3}),
],
)
tag_and_title_query = Q(
"bool",
must=[
Q("match", title={"query": search_param, "fuzziness": "AUTO"}),
Q("match", tags__name={"query": search_param, "fuzziness": "AUTO", "boost": 2.5}),
],
)
category_query = Q("match", categories__name={"query": search_param, "fuzziness": "AUTO", "boost": 2})
tag_query = Q("match", tags__name={"query": search_param, "fuzziness": "AUTO", "boost": 1})
# Combine the queries
combined_query = Q(
"bool",
should=[
category_and_tag_and_title_query,
category_and_title_query,
tag_and_title_query,
title_query,
category_query,
tag_query,
],
)
return combined_query
[docs]
def filter_queryset(self, request, queryset, view):
"""
Filters a QuerySet based on the search parameters in the request.
:param request: HTTP request object.
:param queryset: Initial QuerySet.
:param view: The view that calls this method.
:return: Updated QuerySet.
"""
search_param = request.query_params.get("search")
offset = int(request.query_params.get("offset", 0))
limit = int(request.query_params.get("limit", settings.REST_FRAMEWORK.get("PAGE_SIZE", 1)))
if search_param:
combined_query = self.construct_query(search_param)
search = PostDocument.search().query(combined_query)
total_results = search.count()
if offset >= total_results:
return queryset.none()
end = offset + limit
search = search[offset:end]
queryset = search.to_queryset()
queryset.search_count = total_results
return queryset