107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
from rest_framework import decorators, generics, permissions, response, status, viewsets
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from apps.accounts.models import Customer, User
|
|
from apps.accounts.serializers import CustomerSerializer, UserSerializer
|
|
from apps.core.errors import error_response
|
|
from apps.rbac.permissions import ActionPermission
|
|
|
|
|
|
class UserViewSet(viewsets.ModelViewSet):
|
|
queryset = User.objects.prefetch_related("user_roles__role").all()
|
|
serializer_class = UserSerializer
|
|
permission_classes = [ActionPermission]
|
|
code_map = {
|
|
"list": "user.view",
|
|
"retrieve": "user.view",
|
|
"create": "user.manage",
|
|
"update": "user.manage",
|
|
"partial_update": "user.manage",
|
|
"destroy": "user.manage",
|
|
"set_password": "user.manage",
|
|
}
|
|
|
|
def perform_create(self, serializer):
|
|
password = self.request.data.get("password") or "changeme123"
|
|
user = serializer.save()
|
|
user.set_password(password)
|
|
user.save(update_fields=["password"])
|
|
self._sync_roles(user)
|
|
|
|
def perform_update(self, serializer):
|
|
user = serializer.save()
|
|
self._sync_roles(user)
|
|
|
|
def _sync_roles(self, user):
|
|
role_ids = self.request.data.get("role_ids")
|
|
if role_ids is None:
|
|
return
|
|
from apps.rbac.models import Role, UserRole
|
|
|
|
user.user_roles.all().delete()
|
|
for role_id in role_ids:
|
|
role = Role.objects.filter(id=role_id).first()
|
|
if role is not None:
|
|
UserRole.objects.create(user=user, role=role)
|
|
|
|
@decorators.action(detail=True, methods=["post"], url_path="set-password")
|
|
def set_password(self, request, pk=None):
|
|
user = self.get_object()
|
|
new_password = request.data.get("new_password", "")
|
|
if len(new_password) < 6:
|
|
return Response(
|
|
error_response("WEAK_PASSWORD", "Password must be at least 6 characters."),
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
user.set_password(new_password)
|
|
user.save(update_fields=["password"])
|
|
return Response({"success": True})
|
|
|
|
|
|
class MeView(APIView):
|
|
"""View and update the authenticated user's own profile."""
|
|
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
|
|
def get(self, request):
|
|
return Response(UserSerializer(request.user).data)
|
|
|
|
def patch(self, request):
|
|
user = request.user
|
|
for field in ("first_name", "last_name", "email", "phone"):
|
|
if field in request.data:
|
|
setattr(user, field, request.data[field])
|
|
user.save()
|
|
return Response(UserSerializer(user).data)
|
|
|
|
|
|
class ChangePasswordView(APIView):
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
|
|
def post(self, request):
|
|
old_password = request.data.get("old_password", "")
|
|
new_password = request.data.get("new_password", "")
|
|
if not request.user.check_password(old_password):
|
|
return Response(
|
|
error_response("INVALID_PASSWORD", "Current password is incorrect."),
|
|
status=400,
|
|
)
|
|
if len(new_password) < 6:
|
|
return Response(
|
|
error_response("WEAK_PASSWORD", "New password must be at least 6 characters."),
|
|
status=400,
|
|
)
|
|
request.user.set_password(new_password)
|
|
request.user.save(update_fields=["password"])
|
|
return Response({"success": True})
|
|
|
|
|
|
class CustomerProfileView(generics.RetrieveAPIView):
|
|
permission_classes = [permissions.IsAuthenticated]
|
|
serializer_class = CustomerSerializer
|
|
|
|
def get_object(self):
|
|
customer, _ = Customer.objects.get_or_create(user=self.request.user)
|
|
return customer
|