-
Notifications
You must be signed in to change notification settings - Fork 2.2k
/
api.py
237 lines (207 loc) · 8.88 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
"""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
import drf_yasg.openapi as openapi
from drf_yasg.utils import swagger_auto_schema
from django.utils.decorators import method_decorator
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.parsers import FormParser, JSONParser, MultiPartParser
from rest_framework.authtoken.models import Token
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework import generics
from rest_framework.permissions import IsAuthenticated
from rest_framework.exceptions import MethodNotAllowed
from core.permissions import all_permissions, ViewClassPermission
from users.models import User
from users.serializers import UserSerializer, UserSerializerUpdate
from users.functions import check_avatar
logger = logging.getLogger(__name__)
@method_decorator(name='update', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='Save user details',
operation_description="""
Save details for a specific user, such as their name or contact information, in Label Studio.
""",
manual_parameters=[
openapi.Parameter(
name='id',
type=openapi.TYPE_INTEGER,
in_=openapi.IN_PATH,
description='User ID'),
],
request_body=UserSerializer
))
@method_decorator(name='list', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='List users',
operation_description='List the users that exist on the Label Studio server.'
))
@method_decorator(name='create', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='Create new user',
operation_description='Create a user in Label Studio.',
request_body=UserSerializer
))
@method_decorator(name='retrieve', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='Get user info',
operation_description='Get info about a specific Label Studio user, based on the user ID.',
manual_parameters = [
openapi.Parameter(
name='id',
type=openapi.TYPE_INTEGER,
in_=openapi.IN_PATH,
description='User ID'),
],
))
@method_decorator(name='partial_update', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='Update user details',
operation_description="""
Update details for a specific user, such as their name or contact information, in Label Studio.
""",
manual_parameters=[
openapi.Parameter(
name='id',
type=openapi.TYPE_INTEGER,
in_=openapi.IN_PATH,
description='User ID'),
],
request_body=UserSerializer
))
@method_decorator(name='destroy', decorator=swagger_auto_schema(
tags=['Users'],
operation_summary='Delete user',
operation_description='Delete a specific Label Studio user.',
manual_parameters=[
openapi.Parameter(
name='id',
type=openapi.TYPE_INTEGER,
in_=openapi.IN_PATH,
description='User ID'),
],
))
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)
@swagger_auto_schema(auto_schema=None, methods=['delete', 'post'])
@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 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=swagger_auto_schema(
tags=['Users'],
operation_summary='Reset user token',
operation_description='Reset the user token for the current user.',
responses={
201: openapi.Response(
description='User token response',
schema=openapi.Schema(
description='User token',
type=openapi.TYPE_OBJECT,
properties={
'token': openapi.Schema(
description='Token',
type=openapi.TYPE_STRING
)
}
))
}))
class UserResetTokenAPI(APIView):
parser_classes = (JSONParser, FormParser, MultiPartParser)
queryset = User.objects.all()
permission_classes = (IsAuthenticated,)
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=swagger_auto_schema(
tags=['Users'],
operation_summary='Get user token',
operation_description='Get a user token to authenticate to the API as the current user.',
responses={
200: openapi.Response(
description='User token response',
type=openapi.TYPE_OBJECT,
properties={
'detail': openapi.Schema(
description='Token',
type=openapi.TYPE_STRING
)
}
)
}))
class UserGetTokenAPI(APIView):
parser_classes = (JSONParser,)
permission_classes = (IsAuthenticated,)
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=swagger_auto_schema(
tags=['Users'],
operation_summary='Retrieve my user',
operation_description='Retrieve details of the account that you are using to access the API.'
))
class UserWhoAmIAPI(generics.RetrieveAPIView):
parser_classes = (JSONParser, FormParser, MultiPartParser)
queryset = User.objects.all()
permission_classes = (IsAuthenticated, )
serializer_class = UserSerializer
def get_object(self):
return self.request.user
def get(self, request, *args, **kwargs):
return super(UserWhoAmIAPI, self).get(request, *args, **kwargs)