"""This file and its contents are licensed under the Apache License 2.0. Please see the included NOTICE for copyright information and LICENSE for a copy of the license. """ import logging from core.permissions import ViewClassPermission, all_permissions from django.utils.decorators import method_decorator from drf_spectacular.types import OpenApiTypes from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema from rest_framework import generics, viewsets from rest_framework.authtoken.models import Token from rest_framework.decorators import action from rest_framework.exceptions import MethodNotAllowed from rest_framework.parsers import FormParser, JSONParser, MultiPartParser from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from users.functions import check_avatar from users.models import User from users.serializers import HotkeysSerializer, UserSerializer, UserSerializerUpdate, WhoAmIUserSerializer logger = logging.getLogger(__name__) _user_schema = { 'type': 'object', 'properties': { 'id': { 'type': 'integer', 'description': 'User ID', }, 'first_name': { 'type': 'string', 'description': 'First name of the user', }, 'last_name': { 'type': 'string', 'description': 'Last name of the user', }, 'username': { 'type': 'string', 'description': 'Username of the user', }, 'email': { 'type': 'string', 'description': 'Email of the user', }, 'avatar': { 'type': 'string', 'description': 'Avatar URL of the user', }, 'initials': { 'type': 'string', 'description': 'Initials of the user', }, 'phone': { 'type': 'string', 'description': 'Phone number of the user', }, 'allow_newsletters': { 'type': 'boolean', 'description': 'Whether the user allows newsletters', }, }, } @method_decorator( name='update', decorator=extend_schema( tags=['Users'], summary='Save user details', description=""" Save details for a specific user, such as their name or contact information, in Label Studio. """, parameters=[ OpenApiParameter(name='id', type=OpenApiTypes.INT, location='path', description='User ID'), ], request=UserSerializer, extensions={ 'x-fern-audiences': ['internal'], }, ), ) @method_decorator( name='list', decorator=extend_schema( tags=['Users'], summary='List users', description='List the users that exist on the Label Studio server.', extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'list', 'x-fern-audiences': ['public'], }, ), ) @method_decorator( name='create', decorator=extend_schema( tags=['Users'], summary='Create new user', description='Create a user in Label Studio.', request={ 'application/json': _user_schema, }, responses={201: UserSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'create', 'x-fern-audiences': ['public'], }, ), ) @method_decorator( name='retrieve', decorator=extend_schema( tags=['Users'], summary='Get user info', description='Get info about a specific Label Studio user, based on the user ID.', parameters=[ OpenApiParameter(name='id', type=OpenApiTypes.INT, location='path', description='User ID'), ], request=None, responses={200: UserSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'get', 'x-fern-audiences': ['public'], }, ), ) @method_decorator( name='partial_update', decorator=extend_schema( tags=['Users'], summary='Update user details', description=""" Update details for a specific user, such as their name or contact information, in Label Studio. """, parameters=[ OpenApiParameter(name='id', type=OpenApiTypes.INT, location='path', description='User ID'), ], request={ 'application/json': _user_schema, }, responses={200: UserSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'update', 'x-fern-audiences': ['public'], }, ), ) @method_decorator( name='destroy', decorator=extend_schema( tags=['Users'], summary='Delete user', description='Delete a specific Label Studio user.', parameters=[ OpenApiParameter(name='id', type=OpenApiTypes.INT, location='path', description='User ID'), ], request=None, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'delete', 'x-fern-audiences': ['public'], }, ), ) class UserAPI(viewsets.ModelViewSet): serializer_class = UserSerializer permission_required = ViewClassPermission( GET=all_permissions.organizations_change, PUT=all_permissions.organizations_change, POST=all_permissions.organizations_change, PATCH=all_permissions.organizations_view, DELETE=all_permissions.organizations_change, ) http_method_names = ['get', 'post', 'head', 'patch', 'delete'] def get_queryset(self): return User.objects.filter(organizations=self.request.user.active_organization) @extend_schema(exclude=True) @action(detail=True, methods=['delete', 'post'], permission_required=all_permissions.avatar_any) def avatar(self, request, pk): if request.method == 'POST': avatar = check_avatar(request.FILES) request.user.avatar = avatar request.user.save() return Response({'detail': 'avatar saved'}, status=200) elif request.method == 'DELETE': request.user.avatar = None request.user.save() return Response(status=204) def get_serializer_class(self): if self.request.method in {'PUT', 'PATCH'}: return UserSerializerUpdate return super().get_serializer_class() def get_serializer_context(self): context = super(UserAPI, self).get_serializer_context() context['user'] = self.request.user return context def update(self, request, *args, **kwargs): return super(UserAPI, self).update(request, *args, **kwargs) def list(self, request, *args, **kwargs): return super(UserAPI, self).list(request, *args, **kwargs) def create(self, request, *args, **kwargs): return super(UserAPI, self).create(request, *args, **kwargs) def perform_create(self, serializer): instance = serializer.save() self.request.user.active_organization.add_user(instance) def retrieve(self, request, *args, **kwargs): return super(UserAPI, self).retrieve(request, *args, **kwargs) def partial_update(self, request, *args, **kwargs): result = super(UserAPI, self).partial_update(request, *args, **kwargs) # throw MethodNotAllowed if read-only fields are attempted to be updated read_only_fields = self.get_serializer_class().Meta.read_only_fields for field in read_only_fields: if field in request.data: raise MethodNotAllowed('PATCH', detail=f'Cannot update read-only field: {field}') # newsletters if 'allow_newsletters' in request.data: user = User.objects.get(id=request.user.id) # we need an updated user request.user.advanced_json = { # request.user instance will be unchanged in request all the time 'email': user.email, 'allow_newsletters': user.allow_newsletters, 'update-notifications': 1, 'new-user': 0, } return result def destroy(self, request, *args, **kwargs): return super(UserAPI, self).destroy(request, *args, **kwargs) @method_decorator( name='post', decorator=extend_schema( tags=['Users'], summary='Reset user token', description='Reset the user token for the current user.', request=None, responses={ 201: OpenApiResponse( description='User token response', response={ 'type': 'object', 'properties': {'token': {'type': 'string'}}, }, ) }, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'reset_token', 'x-fern-audiences': ['public'], }, ), ) class UserResetTokenAPI(APIView): parser_classes = (JSONParser, FormParser, MultiPartParser) queryset = User.objects.all() permission_required = all_permissions.users_token_any def post(self, request, *args, **kwargs): user = request.user token = user.reset_token() logger.debug(f'New token for user {user.pk} is {token.key}') return Response({'token': token.key}, status=201) @method_decorator( name='get', decorator=extend_schema( tags=['Users'], summary='Get user token', description='Get a user token to authenticate to the API as the current user.', request=None, responses={ 200: OpenApiResponse( description='User token response', response={ 'type': 'object', 'properties': {'detail': {'type': 'string'}}, }, ) }, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'get_token', 'x-fern-audiences': ['public'], }, ), ) class UserGetTokenAPI(APIView): parser_classes = (JSONParser,) permission_required = all_permissions.users_token_any def get(self, request, *args, **kwargs): user = request.user token = Token.objects.get(user=user) return Response({'token': str(token)}, status=200) @method_decorator( name='get', decorator=extend_schema( tags=['Users'], summary='Retrieve my user', description='Retrieve details of the account that you are using to access the API.', request=None, responses={200: WhoAmIUserSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'whoami', 'x-fern-audiences': ['public'], }, ), ) class UserWhoAmIAPI(generics.RetrieveAPIView): parser_classes = (JSONParser, FormParser, MultiPartParser) queryset = User.objects.all() permission_classes = (IsAuthenticated,) serializer_class = WhoAmIUserSerializer def get_object(self): return self.request.user def get(self, request, *args, **kwargs): return super(UserWhoAmIAPI, self).get(request, *args, **kwargs) @method_decorator( name='patch', decorator=extend_schema( tags=['Users'], summary='Update user hotkeys', description='Update the custom hotkeys configuration for the current user.', request=HotkeysSerializer, responses={200: HotkeysSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'update_hotkeys', 'x-fern-audiences': ['public'], }, ), ) @method_decorator( name='get', decorator=extend_schema( tags=['Users'], summary='Get user hotkeys', description='Retrieve the custom hotkeys configuration for the current user.', request=None, responses={200: HotkeysSerializer}, extensions={ 'x-fern-sdk-group-name': 'users', 'x-fern-sdk-method-name': 'get_hotkeys', 'x-fern-audiences': ['public'], }, ), ) class UserHotkeysAPI(APIView): permission_classes = [IsAuthenticated] parser_classes = (JSONParser, FormParser, MultiPartParser) def get(self, request, *args, **kwargs): """Retrieve the current user's hotkeys configuration""" try: user = request.user custom_hotkeys = user.custom_hotkeys or {} serializer = HotkeysSerializer(data={'custom_hotkeys': custom_hotkeys}) if serializer.is_valid(): return Response(serializer.validated_data, status=200) else: # If stored data is invalid, return empty config logger.warning(f'Invalid hotkeys data for user {user.pk}: {serializer.errors}') return Response({'custom_hotkeys': {}}, status=200) except Exception as e: logger.error(f'Error retrieving hotkeys for user {request.user.pk}: {str(e)}') return Response({'error': 'Failed to retrieve hotkeys configuration'}, status=500) def patch(self, request, *args, **kwargs): """Update the current user's hotkeys configuration""" try: serializer = HotkeysSerializer(data=request.data) if not serializer.is_valid(): return Response({'error': 'Invalid hotkeys configuration', 'details': serializer.errors}, status=400) user = request.user # Security check: Ensure user can only update their own hotkeys if not user.is_authenticated: return Response({'error': 'Authentication required'}, status=401) # Update user's hotkeys user.custom_hotkeys = serializer.validated_data['custom_hotkeys'] user.save(update_fields=['custom_hotkeys']) logger.info(f'Updated hotkeys for user {user.pk}') return Response(serializer.validated_data, status=200) except Exception as e: logger.error(f'Error updating hotkeys for user {request.user.pk}: {str(e)}') return Response({'error': 'Failed to update hotkeys configuration'}, status=500)