# apps/users/views/user_viewset.py from rest_framework.decorators import action from rest_framework.response import Response from rest_framework import status, throttling, filters from django_filters.rest_framework import DjangoFilterBackend from django.utils import timezone from rest_framework_simplejwt.tokens import RefreshToken from rest_framework.permissions import AllowAny from django.db import transaction from apps.core.views.base import BaseViewSet from apps.core.utils import create_dynamic_pagination from apps.core.permissions.user_permission import UserPermissionCheck from apps.users.models import User from apps.users.serializers.user import ( UserSerializer, LoginSerializer, UserPermissionsUpdateSerializer, UserRoleChangeSerializer ) from apps.users.serializers.login import EmailTokenObtainPairSerializer from apps.users.serializers.logout import LogoutSerializer class UserViewSet(BaseViewSet): queryset = User.objects.all() serializer_class = UserSerializer permission_classes = [UserPermissionCheck] pagination_class = create_dynamic_pagination(10) throttle_classes = [throttling.ScopedRateThrottle] throttle_scope = 'custom_scope_one' permission_app_name = 'users' permission_model_name = 'user' view_permissions = ["list", "create", "update", "partial_update", "retrieve", "destroy", "manage_permissions", "toggle_status", 'change_role', 'reset_password'] filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['role', 'is_active', 'is_email_verified'] search_fields = ['name', 'email', 'refer_code'] ordering_fields = ['name', 'date_joined'] def get_serializer_class(self): action_serializer_map = { "manage_permissions": UserPermissionsUpdateSerializer, "change_role": UserRoleChangeSerializer, "login": EmailTokenObtainPairSerializer, "logout": LogoutSerializer, "register": UserSerializer, } return action_serializer_map.get(self.action, UserSerializer) def get_queryset(self): try: user = self.request.user if not user.is_authenticated: return User.objects.none() queryset = User.objects.all() if user.is_superuser or user.role == 'admin': return queryset return queryset.filter(id=user.id) except Exception: return User.objects.none() @action(detail=False, methods=['post'], permission_classes=[AllowAny]) def register(self, request): try: incoming_refer_code = request.query_params.get('ref') serializer = self.get_serializer(data=request.data) if not serializer.is_valid(): return Response({ "success": False, "message": "Registration failed.", "errors": serializer.errors }, status=status.HTTP_400_BAD_REQUEST) with transaction.atomic(): user = serializer.save() if incoming_refer_code: try: if user.refer_code != incoming_refer_code: referrer = User.objects.get(refer_code=incoming_refer_code) from apps.profiles.models import Profile profile, created = Profile.objects.get_or_create(user=user) profile.super_refer_code = referrer profile.is_referred = True profile.save() except User.DoesNotExist: pass return Response({ "success": True, "message": "Registration successful.", "data": { "id": str(user.id), "email": user.email, "refer_code": user.refer_code } }, status=status.HTTP_201_CREATED) except Exception as e: return Response({ "success": False, "message": "An unexpected error occurred." }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @action(detail=False, methods=['post'], permission_classes=[AllowAny]) def login(self, request): try: login_serializer = LoginSerializer(data=request.data) if not login_serializer.is_valid(): return Response({"success": False, "errors": login_serializer.errors}, status=status.HTTP_400_BAD_REQUEST) token_serializer = EmailTokenObtainPairSerializer(data=request.data, context={'request': request}) token_serializer.is_valid(raise_exception=True) user = User.objects.get(email=request.data.get('email')) user.last_login = timezone.now() user.save(update_fields=['last_login']) return Response(token_serializer.validated_data, status=status.HTTP_200_OK) except Exception as e: return Response({"success": False, "message": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @action(detail=True, methods=['get', 'post']) def manage_permissions(self, request, pk=None): try: user_to_update = self.get_object() if request.method == 'GET': return Response({ "success": True, "data": user_to_update.permissions }, status=status.HTTP_200_OK) if user_to_update == request.user: return Response({"success": False, "message": "Cannot change own permissions."}, status=status.HTTP_403_FORBIDDEN) serializer = self.get_serializer(data=request.data) if not serializer.is_valid(): return Response({"success": False, "errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST) user_to_update.permissions = serializer.validated_data['permissions'] user_to_update.save() return Response({"success": True, "message": "Permissions updated."}, status=status.HTTP_200_OK) except Exception as e: return Response({"success": False, "message": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @action(detail=True, methods=['get','post']) def change_role(self, request, pk=None): try: user = self.get_object() serializer = self.get_serializer(data=request.data, context={'request': request, 'user': user}) if not serializer.is_valid(): return Response({"success": False, "errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST) with transaction.atomic(): new_role = serializer.validated_data['role'] user.role = new_role from apps.core.permissions.role_permission import get_role_based_permissions user.permissions = get_role_based_permissions(new_role) user.save() return Response({"success": True, "message": f"Role updated to {new_role}"}, status=status.HTTP_200_OK) except Exception as e: return Response({"success": False, "message": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) @action(detail=True, methods=['post']) def reset_password(self, request, pk=None): try: user = self.get_object() if not request.user.is_superuser and user != request.user: return Response({"success": False, "message": "Unauthorized."}, status=status.HTTP_403_FORBIDDEN) new_pass = request.data.get('new_password') confirm_pass = request.data.get('confirm_password') if new_pass != confirm_pass: return Response({"success": False, "message": "Passwords mismatch."}, status=status.HTTP_400_BAD_REQUEST) if not request.user.is_superuser: current_pass = request.data.get('current_password') if not current_pass or not user.check_password(current_pass): return Response({"success": False, "message": "Invalid current password."}, status=status.HTTP_400_BAD_REQUEST) user.set_password(new_pass) user.save() return Response({"success": True, "message": "Password reset successful."}, status=status.HTTP_200_OK) except Exception as e: return Response({"success": False, "message": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR) def create(self, request, *args, **kwargs): try: if self.action == 'register': return self.register(request) if not request.user.is_superuser: return Response({"success": False, "message": "Admin access only."}, status=status.HTTP_403_FORBIDDEN) return super().create(request, *args, **kwargs) except Exception as e: return Response({"success": False, "message": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)