from dataclasses import asdict
from typing import Any, Dict, Optional, Tuple, cast
from django.conf import settings
from django.contrib.auth.base_user import AbstractBaseUser
from django.contrib.auth.models import AnonymousUser
from django.db.models import BooleanField, Count, Exists, OuterRef, QuerySet, Subquery
from django.http import Http404, HttpResponse
from rest_framework import permissions, status
from rest_framework.decorators import action
from rest_framework.filters import SearchFilter
from rest_framework.generics import ListAPIView
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet
from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
from rest_framework_simplejwt.tokens import AccessToken
from api.posts.pagination import SearchPagination
from api.users.docs import (
activate_docs,
confirm_email_docs,
confirm_password_docs,
interest_list_docs,
password_change_docs,
resend_activation_code_docs,
reset_password_docs,
signup_docs,
token_create_docs,
token_refresh_docs,
users_list_docs,
)
from api.users.exceptions import AlreadyFollowingUserAPIException, NotFollowingUserAPIException
from api.users.filters import UserElasticsearchFilter
from api.users.serializers import (
CategorySerializer,
ConfirmEmailSerializer,
ConfirmPasswordSerializer,
CustomTokenObtainPairSerializer,
CustomTokenRefreshSerializer,
ExtendedUserSerializer,
FollowingUserSerializer,
ResendUserActivationCodeSerializer,
UserActivationSerializer,
UserChangePasswordSerializer,
UserResetPassword,
UserSerializer,
)
from apps.users.models import Category, User
from common.views import CustomModelViewSet, CustomViewSet, action_with_serializer # type: ignore
from services.users import UsersService
from services.users.exceptions import AlreadyFollowingUserException, NotFollowingUserException
from services.users.export import ExportUserData
from services.users.unauthorized import UnauthorizedUserService
from tasks.users import send_password_activation
[docs]
class InterestsListAPIView(ListAPIView):
"""
View with interests list
"""
serializer_class = CategorySerializer
queryset = Category.objects.all().order_by("id")
http_method_names = ["get"]
tags = ["Users [unauthorized]"]
[docs]
@interest_list_docs
def get(self, request, *args, **kwargs):
return super().get(request, *args, **kwargs)
[docs]
class UsersUnauthorizedViewSet(CustomViewSet):
"""
ViewSet with actions for unauthorized users
"""
permission_classes = [permissions.AllowAny]
http_method_names = ["post"]
tags = ["Users [unauthorized]"]
service = UsersService
[docs]
@signup_docs
@action_with_serializer(
detail=False, methods=["post"], serializer_class=UserSerializer, permission_classes=[permissions.AllowAny]
)
def signup(self, request: Request, *args: Tuple[Any], **kwargs: Dict) -> Response:
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
profile_data = data.pop("profile", None)
interests_data = data.pop("interests", None)
password = data.pop("password")
# Delete password_repeat field because it was needed only for validation
data.pop("password_repeat")
instance = UnauthorizedUserService.create_user(
data=data, profile=profile_data, interests=interests_data, password=password
)
serializer = self.get_serializer(instance)
return Response(data=serializer.data, status=status.HTTP_200_OK)
[docs]
@activate_docs
@action_with_serializer(methods=["POST"], detail=True, serializer_class=UserActivationSerializer)
def activate(self, request, *args: Tuple[Any], **kwargs: Dict):
data = {"uid": kwargs["uid"], "token": kwargs["token"]}
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
response_data = self.service(serializer.validated_data.get("user")).activate()
return Response(data=asdict(response_data), status=status.HTTP_200_OK)
[docs]
@reset_password_docs
@action_with_serializer(methods=["POST"], detail=True, serializer_class=UserResetPassword)
def reset_password(self, request, *args: Tuple[Any], **kwargs: Dict):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
if settings.IS_TESTING:
send_password_activation(user_email=serializer.validated_data["email"])
else:
send_password_activation.delay(user_email=serializer.validated_data["email"])
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@confirm_password_docs
@action_with_serializer(methods=["POST"], detail=True, serializer_class=ConfirmPasswordSerializer)
def confirm_password(self, request, *args: Tuple[Any], **kwargs: Dict):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
self.service(data["user"]).confirm_password(password=data["password"], uid=data["uid"], token=data["token"])
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@confirm_email_docs
@action_with_serializer(methods=["POST"], detail=True, serializer_class=ConfirmEmailSerializer)
def confirm_email(self, request, *args: Tuple[Any], **kwargs: Dict):
data: Dict = {"uid": kwargs["uid"], "token": kwargs["token"]}
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
self.service(serializer.validated_data["user"]).confirm_email(uid=data["uid"], token=data["token"])
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@resend_activation_code_docs
@action_with_serializer(methods=["POST"], detail=True, serializer_class=ResendUserActivationCodeSerializer)
def resend_activation_code(self, request, *args: Tuple[Any], **kwargs: Dict):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = User.objects.get(email=serializer.validated_data.get("email"))
self.service(user).send_activation_email(is_created=False)
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
class UsersAuthorizedViewSet(CustomModelViewSet):
tags = ["Users [authorized]"]
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
http_method_names = ["post", "get", "put", "patch", "delete"]
serializer_class = ExtendedUserSerializer
filter_backends = [UserElasticsearchFilter]
pagination_class = SearchPagination
service = UsersService
[docs]
def get_queryset(self) -> QuerySet[User]:
likes = User.objects.filter(id=OuterRef("id")).annotate(like_count=Count("liked_posts")).values("like_count")
# subquery for count followers
followers = (
User.objects.filter(id=OuterRef("id")).annotate(follower_count=Count("followers")).values("follower_count")
)
# subquery for count following users
following = (
User.objects.filter(id=OuterRef("id"))
.annotate(following_count=Count("following"))
.values("following_count")
)
annotations = {
"likes_quantity": Subquery(likes[:1]),
"followers_quantity": Subquery(followers[:1]),
"following_quantity": Subquery(following[:1]),
}
# Add is_current_user_following to annotations if the user is authenticated
if self.request.user.is_authenticated:
user = cast(User, self.request.user)
is_following_subquery = Exists(user.following.filter(pk=OuterRef("pk")))
annotations["is_current_user_following"] = is_following_subquery
queryset = User.objects.annotate(**annotations).order_by("-followers_quantity")
return queryset
[docs]
def get_object(self) -> Optional[AbstractBaseUser | AnonymousUser]:
if self.kwargs.get("pk"):
pk = self.kwargs["pk"]
else:
pk = self.request.user.pk
try:
user = self.get_queryset().get(pk=pk, is_deleted=False)
except User.DoesNotExist:
raise Http404("User not found")
return user
[docs]
@password_change_docs
@action_with_serializer(
detail=False,
methods=["post"],
serializer_class=UserChangePasswordSerializer,
permission_classes=[permissions.IsAuthenticated],
)
def password_change(self, request: Request):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
if not isinstance(request.user, AnonymousUser):
user = request.user
self.service(user).set_password(serializer.validated_data.get("password"))
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@action(detail=False, methods=["post"])
def export_data(self, request):
user_data = ExportUserData.export_data(request.user)
response = HttpResponse(user_data, content_type="application/json", status=200)
response["Content-Disposition"] = 'attachment; filename="user_data.json"'
return response
[docs]
@users_list_docs
def list(self, request: Request, *args: Any, **kwargs: Any) -> Response:
return super(UsersAuthorizedViewSet, self).list(request, args, kwargs)
[docs]
class TokenViewSet(CustomViewSet):
permission_classes = [permissions.AllowAny]
http_method_names = ["post"]
tags = ["Token"]
[docs]
def handle_token_response(self, serializer_class):
serializer = serializer_class(data=self.request.data)
serializer.is_valid(raise_exception=True)
token = serializer.validated_data["access"]
try:
# AccessToken is a special class that can be used to decode and handle access tokens.
token = AccessToken(token)
except TokenError:
raise InvalidToken("This token is invalid")
user_pk = token["user_id"]
user = User.objects.prefetch_related("following").get(pk=user_pk)
data = {"user": FollowingUserSerializer(user, is_owner=True).data}
data.update(serializer.validated_data)
if user.is_deleted:
raise Http404("User not found")
return Response(data, status=status.HTTP_200_OK)
[docs]
@token_create_docs
@action(detail=False, methods=["post"], serializer_class=CustomTokenObtainPairSerializer)
def obtain(self, request):
return self.handle_token_response(CustomTokenObtainPairSerializer)
[docs]
@token_refresh_docs
@action(detail=False, methods=["post"], serializer_class=CustomTokenRefreshSerializer)
def refresh(self, request):
return self.handle_token_response(CustomTokenRefreshSerializer)
[docs]
class FollowersAPIViewSet(GenericViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
service = UsersService
http_method_names = ["get", "post"]
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
filter_backends = [SearchFilter]
search_fields = ["username", "first_name", "last_name"]
tags = ["Users [authorized]"]
subtag = "followers"
[docs]
def list(self, request, return_followers: bool = True, *args, **kwargs):
queryset = self.filter_queryset(self._get_followers_queryset(return_followers))
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
def _get_followers_queryset(self, return_followers: bool = True) -> "QuerySet[User]":
"""
Returns the queryset of followers for the authenticated user.
:return_followers: returns followers users otherwise returns following users
Returns:
QuerySet[User]: The queryset of followers.
"""
user = self.get_object()
# Check if the user exists
if not user:
return User.objects.none()
if return_followers:
# Return followers of the user
queryset = user.followers.all()
else:
# Return users the user is following
queryset = user.following.all()
# Create a subquery that checks if the user in the main query is followed by the request.user
current_user = cast(User, self.request.user)
if current_user.is_authenticated:
is_following_subquery = current_user.following.filter(pk=OuterRef("pk"))
# Annotate the main query with the is_current_user_following field using the subquery
queryset = queryset.annotate(
is_current_user_following=Exists(is_following_subquery, output_field=BooleanField())
)
return queryset
[docs]
def get_object(self) -> Optional[User]:
if "pk" in self.kwargs:
try:
return User.objects.get(pk=self.kwargs.get("pk"))
except User.DoesNotExist:
raise Http404
else:
if self.request.user.is_authenticated:
return self.request.user
else:
return None
[docs]
@action(detail=True, methods=["post"])
def follow(self, request, pk=None):
following_user = self.get_object()
try:
self.service(request.user).follow(following_user)
except AlreadyFollowingUserException:
raise AlreadyFollowingUserAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@action(detail=True, methods=["post"])
def unfollow(self, request, pk=None):
following_user = self.get_object()
try:
self.service(request.user).unfollow(following_user)
except NotFollowingUserException:
raise NotFollowingUserAPIException()
return Response(status=status.HTTP_204_NO_CONTENT)
[docs]
@users_list_docs
@action(detail=False, methods=["get"])
def followers_list(self, request, pk=None):
return self.list(request, return_followers=True)
[docs]
@users_list_docs
@action(detail=False, methods=["get"])
def following_list(self, request, pk=None):
return self.list(request, return_followers=False)