from typing import Any, cast
from django.db.models import (
BooleanField,
Case,
Exists,
ExpressionWrapper,
F,
FloatField,
OuterRef,
Q,
QuerySet,
Value,
When,
)
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.generics import CreateAPIView, GenericAPIView, ListAPIView, RetrieveUpdateDestroyAPIView
from rest_framework.mixins import ListModelMixin
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import ModelViewSet
from api.posts.docs import (
add_post_to_favorites_docs,
favorites_list_docs,
list_posts_docs,
remove_post_from_favorites_docs,
)
from api.posts.exceptions import (
AlreadyAddedToFavoritesAPIException,
AlreadyLikedPostAPIException,
UserNotHasThisPostInFavoritesAPIException,
UserNotLikeThisPostAPIException,
)
from api.posts.filters import ElasticsearchFilter, PostFilter
from api.posts.pagination import SearchPagination
from api.posts.permissions import IsImageRelatedToCommentPermission, IsOwnerOrReadOnly
from api.posts.serializers import (
CommentImageSerilizer,
CommentListSerializer,
CommentSerializer,
FavoriteSerializer,
PostContentsSerializer,
PostMediaSerializer,
PostSerializer,
UpdateCommentSerializer,
WebhookPostSerializer,
)
from apps.posts.models import Comment, CommentImage, Favorite, Like, Post, PostMedia
from apps.users.models import User
from common.views import CustomGenericViewSet, CustomModelViewSet # type: ignore
from services.posts import AlreadyLikedPostException, PostCreationService, PostService, UserNotLikeThisPostException
from services.posts.comment import CommentService
from services.posts.comment_image import CommentImageService
from services.posts.exceptions import AlreadyAddedToFavoritesException, UserNotHasThisPostInFavoritesException
[docs]
class PostAPIViewSet(CustomModelViewSet):
serializer_class = PostSerializer
permission_classes = [IsOwnerOrReadOnly]
service_class = PostService
http_method_names = ["get", "post", "put", "patch", "delete"]
tags = ["Posts"]
filter_backends = [DjangoFilterBackend, ElasticsearchFilter]
filterset_class = PostFilter
pagination_class = SearchPagination
[docs]
def perform_destroy(self, instance: Post) -> None:
self.service_class(instance).delete()
def _get_explore_queryset(self) -> QuerySet[Post]:
"""
Returns a queryset with personalized posts (based on likes and views)
"""
liked_subquery = Like.objects.filter(author=self.request.user, post=OuterRef("pk"))
favorite_subquery = Favorite.objects.filter(author=self.request.user, post=OuterRef("pk"))
safe_views = Case(When(views_quantity=0, then=1), default=F("views_quantity"), output_field=FloatField())
is_following_subquery = self.request.user.following.filter(pk=OuterRef("author_id")) # type: ignore
likes_to_views_weight = 1
likes_weight = 1
interest_weight = 1
queryset = (
Post.objects.visible_for_user(self.request.user)
.annotate(
is_liked=Exists(liked_subquery),
is_favorite=Exists(favorite_subquery),
is_current_user_following=Exists(is_following_subquery, output_field=BooleanField()),
likes_to_views_ratio=ExpressionWrapper(F("likes_quantity") / safe_views, output_field=FloatField()),
combined_score=ExpressionWrapper(
likes_to_views_weight * F("likes_to_views_ratio")
+ likes_weight * F("likes_quantity")
+ interest_weight,
output_field=FloatField(),
),
)
.order_by("-combined_score")
)
return queryset
def _get_for_you_queryset(self) -> QuerySet[Post]:
"""
Returns a queryset with personalized posts (based on user interests)
"""
user_interests = self.request.user.interests.all()
liked_subquery = Like.objects.filter(author=self.request.user, post=OuterRef("pk"))
favorite_subquery = Favorite.objects.filter(author=self.request.user, post=OuterRef("pk"))
is_following_subquery = self.request.user.following.filter(pk=OuterRef("author_id")) # type: ignore
safe_views = Case(When(views_quantity=0, then=1), default=F("views_quantity"), output_field=FloatField())
likes_to_views_weight = 1
likes_weight = 1
interest_weight = 2
queryset = (
Post.objects.visible_for_user(self.request.user)
.annotate(
is_liked=Exists(liked_subquery),
is_favorite=Exists(favorite_subquery),
is_current_user_following=Exists(is_following_subquery, output_field=BooleanField()),
likes_to_views_ratio=ExpressionWrapper(F("likes_quantity") / safe_views, output_field=FloatField()),
combined_score=ExpressionWrapper(
likes_to_views_weight * F("likes_to_views_ratio")
+ likes_weight * F("likes_quantity")
+ interest_weight
* Case(
When(categories__in=user_interests, then=Value(1)), # TODO: Change logic
default=Value(0),
output_field=FloatField(),
),
output_field=FloatField(),
),
)
.exclude(author=self.request.user)
.order_by("-combined_score")
)
return queryset
def _get_following_queryset(self) -> QuerySet[Post]:
"""
Returns a queryset with personalized posts (based on user following)
"""
liked_subquery = Like.objects.filter(author=self.request.user, post=OuterRef("pk"))
favorite_subquery = Favorite.objects.filter(author=self.request.user, post=OuterRef("pk"))
safe_views = Case(When(views_quantity=0, then=1), default=F("views_quantity"), output_field=FloatField())
is_following_subquery = self.request.user.following.filter(pk=OuterRef("author_id")) # type: ignore
likes_to_views_weight = 1
likes_weight = 1
queryset = (
Post.objects.visible_for_user(self.request.user)
.annotate(
is_liked=Exists(liked_subquery),
is_favorite=Exists(favorite_subquery),
is_current_user_following=Exists(is_following_subquery, output_field=BooleanField()),
likes_to_views_ratio=ExpressionWrapper(F("likes_quantity") / safe_views, output_field=FloatField()),
combined_score=ExpressionWrapper(
likes_to_views_weight * F("likes_to_views_ratio") + likes_weight * F("likes_quantity"),
output_field=FloatField(),
),
)
.order_by("-combined_score")
.filter(author__in=self.request.user.following.all()) # type: ignore
)
return queryset
[docs]
def get_queryset(self) -> QuerySet[Post]: # TODO: separete more
if self.request.user.is_authenticated and (self.action == "list" or self.action == "retrieve"):
queryset = self._get_explore_queryset()
elif self.request.user.is_authenticated and self.action == "for_you":
queryset = self._get_for_you_queryset()
elif self.request.user.is_authenticated and self.action == "following":
queryset = self._get_following_queryset()
elif self.action in ["update", "partial_update"]:
queryset = Post.objects.all()
else:
queryset = Post.objects.public()
return (
queryset.select_related("author", "author__profile")
.prefetch_related("categories", "tags")
.with_comments_count() # type: ignore
.distinct()
)
[docs]
@list_posts_docs
def list(self, request, *args, **kwargs):
return super().list(request, *args, **kwargs)
[docs]
def following(self, request, *args, **kwargs):
return super().list(request, *args, **kwargs)
[docs]
def for_you(self, request, *args, **kwargs):
return super().list(request, *args, **kwargs)
[docs]
def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Response:
"""Retrieve post and add view to post if user didn't see this post before"""
if request.user.is_authenticated:
user = cast(User, request.user)
PostService(self.get_object()).view(user)
return super().retrieve(request, args, kwargs)
[docs]
class UploadPostAPIView(GenericAPIView):
"""Upload post api view"""
queryset = Post.objects.all()
serializer_class = PostContentsSerializer
permission_classes = [IsAuthenticated]
service_class = PostCreationService
[docs]
def post(self, request: Request, *args: Any, **kwargs: Any) -> Response:
"""Upload media with createting post"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
instance = self.service_class.create(serializer.validated_data, self.request.user) # type: ignore
return Response({"post": instance.id}, status=status.HTTP_201_CREATED)
[docs]
class PostLikeAPIView(GenericAPIView):
"""Like post api view"""
queryset = Post.objects.all()
permission_classes = [IsAuthenticated]
service_class = PostService
serializer_class = None
tags = ["Posts"]
subtag = "like"
model_name = "Like"
[docs]
def post(self, request, *args, **kwargs):
"""Post method"""
instance = self.get_object()
try:
PostService(instance).like(request.user)
except AlreadyLikedPostException:
raise AlreadyLikedPostAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
class PostUnLikeAPIView(GenericAPIView):
queryset = Post.objects.all()
permission_classes = [IsAuthenticated]
service_class = PostService
serializer_class = None
tags = ["Posts"]
subtag = "like"
model_name = "Like"
[docs]
def get_serializer_class(self):
pass
[docs]
def post(self, request, *args, **kwargs):
instance = self.get_object()
try:
PostService(instance).unlike(request.user)
except UserNotLikeThisPostException:
raise UserNotLikeThisPostAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
class PostMediaViewSet(ModelViewSet):
http_method_names = ["get", "put", "patch", "delete"]
queryset = PostMedia.objects.all()
permission_classes = [IsOwnerOrReadOnly]
serializer_class = PostMediaSerializer
tags = ["Posts"]
subtag = "media"
model_name = "Media"
[docs]
class FavoriteViewSet(ListModelMixin, CustomGenericViewSet):
http_method_names = ["get", "post", "delete"]
tags = ["Posts"]
permission_classes = [IsAuthenticated]
serializer_class = FavoriteSerializer
subtag = "favorites"
[docs]
def get_queryset(self) -> QuerySet[Favorite]:
is_following_subquery = self.request.user.following.filter(pk=OuterRef("post__author_id"))
return (
Favorite.objects.prefetch_related("post")
.filter(post__is_deleted=False, author=self.request.user)
.annotate(
is_current_user_following=Exists(is_following_subquery, output_field=BooleanField()),
)
.order_by("-created_at")
)
[docs]
def get_post_object(self) -> Post:
if self.request.user.is_authenticated:
queryset = Post.objects.visible_for_user(self.request.user)
else:
queryset = Post.objects.public()
return queryset.get(pk=self.kwargs["pk"])
[docs]
@add_post_to_favorites_docs
@action(methods=["post"], detail=True)
def create(self, request, *args, **kwargs):
instance = self.get_post_object()
try:
PostService(instance).add_to_favorites(request.user)
except AlreadyAddedToFavoritesException:
raise AlreadyAddedToFavoritesAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@remove_post_from_favorites_docs
@action(methods=["delete"], detail=True)
def remove(self, request, *args, **kwargs):
instance = self.get_post_object()
try:
PostService(instance).remove_from_favorites(request.user)
except UserNotHasThisPostInFavoritesException:
raise UserNotHasThisPostInFavoritesAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@favorites_list_docs
def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
return super(FavoriteViewSet, self).list(request, args, kwargs)
[docs]
class WebhookPostMediaAPIView(GenericAPIView):
"""Process webhook api view"""
queryset = PostMedia.objects.all()
serializer_class = WebhookPostSerializer
permission_classes = [AllowAny]
service_class = PostService
# PostService(post_media.post).check_on_publish()
[docs]
def post(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
post = serializer.validated_data.pop("post")
self.service_class(post).set_formatted_path(**serializer.validated_data)
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
class BasePostListAPIView(ListAPIView):
"""
Base ListAPIView for Posts
"""
serializer_class = PostSerializer
permission_classes = [AllowAny]
http_method_names = ["get"]
tags = ["Posts"]
[docs]
class UsersPostsListAPIView(BasePostListAPIView):
"""
List of particular user posts
"""
[docs]
def get_queryset(self) -> QuerySet[Post]:
return Post.objects.public().filter(author_id=self.kwargs.get("pk")).order_by("-created_at")
[docs]
class UsersLikedPostsListAPIView(BasePostListAPIView):
"""
List of users liked posts
"""
permission_classes = [IsAuthenticated]
[docs]
def get_queryset(self) -> QuerySet[Post]:
user = cast(User, self.request.user)
return Post.objects.filter(liked__author=user).order_by("-liked__created_at")
[docs]
class UsersPrivatePostsListAPIView(BasePostListAPIView):
"""
List of users private users posts
"""
permission_classes = [IsAuthenticated]
[docs]
def get_queryset(self) -> QuerySet[Post]:
user = cast(User, self.request.user)
return Post.objects.filter(author=user, visibility=Post.VisibilityChoices.PRIVATE)