forked from itsee-birmingham/django_wce_api
-
Notifications
You must be signed in to change notification settings - Fork 0
/
decorators.py
198 lines (156 loc) · 9.11 KB
/
decorators.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
from django.http import JsonResponse
from django.apps import apps
from django.db.models import Q
from accounts.models import User
from api import views as api_views
# TODO: might want something similar so people can only write to their own data in some models
# this only deals with getting things back since all writing requires login so it not open
def apply_model_get_restrictions(function):
def wrap(request, *args, **kwargs):
target = apps.get_model(kwargs['app'], kwargs['model'])
# first see if we are looking for an item that does not exist
if 'pk' in kwargs:
try:
target.objects.get(pk=kwargs['pk'])
except target.DoesNotExist:
return JsonResponse({'message': "Item does not exist"}, status=404)
# if we get this far we are looking either for a list or a single item which
# does exist (even if permissions mean we can't view it)
try:
availability = target.AVAILABILITY
except AttributeError:
availability = 'private'
if availability is None:
# this is the safest default
availability = 'private'
if availability == 'public':
# open means anyone can read everything - citations data for example
return function(request, *args, **kwargs)
elif availability == 'logged_in':
# anyone logged in can see it
if request.user.is_authenticated:
return function(request, *args, **kwargs)
return JsonResponse({'message': "Authentication required"}, status=401)
elif availability == 'public_or_project':
# anyone can see it if it has a public flag set to True
# if not then only a member of the project or a superuser
# this is for mixed tables like transcriptions and verses
# All hybrid public models need a 'public' entry in the schema
# return server error if not
if 'public' not in target.get_fields() or 'project' not in target.get_fields():
return JsonResponse(
{'message': "Internal server error - model configuation incompatible with API (code 10002)"},
status=500
)
if not request.user.is_authenticated: # we are not logged in
# then you only get the public ones
# assumes a public boolean attribute on the model (which is okay because we have checked above)
query = Q(('public', True))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
if request.user.groups.filter(name='%s_superusers' % kwargs['app']).count() > 0:
return function(request, *args, **kwargs)
if 'project__id' not in request.GET and 'project' not in request.GET:
# if no project specified you can only have the public ones
query = Q(('public', True))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
# Here we need to grab the user fields and add them to the query against the user
project_model = apps.get_model(kwargs['app'], 'Project')
user_fields = project_model.get_user_fields()
query = Q()
query |= Q(('public', True))
for field in user_fields:
query_tuple = api_views.get_query_tuple(user_fields[field], field, request.user)
query |= Q(('project__%s' % (query_tuple[0]), query_tuple[1]))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
elif availability == 'project':
if not request.user.is_authenticated: # we are not logged in
# You get nothing
return JsonResponse({'message': "Authentication required"}, status=401)
if 'project' not in target.get_fields():
return JsonResponse(
{'message': "Internal server error - model configuation incompatible with API (code 10003)"},
status=500
)
# a project must be specified in any request to a model of this type
if 'project__id' not in request.GET and 'project' not in request.GET:
return JsonResponse({'message': "Query not complete - Project must be specified"}, status=400)
if 'project' in request.GET and 'project__id' not in request.GET:
print('WARNING: project should be project__id to make sure this works')
if request.user.groups.filter(name='%s_superusers' % kwargs['app']).count() > 0:
return function(request, *args, **kwargs)
# Here we need to grab the user fields and add them to the query against the user
project_model = apps.get_model(kwargs['app'], 'Project')
user_fields = project_model.get_user_fields()
query = Q()
for field in user_fields:
query_tuple = api_views.get_query_tuple(user_fields[field], field, request.user)
query |= Q(('project__%s' % (query_tuple[0]), query_tuple[1]))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
elif availability == 'project_or_user':
if not request.user.is_authenticated: # we are not logged in
# You get nothing
return JsonResponse({'message': "Authentication required"}, status=401)
if 'project' not in target.get_fields():
return JsonResponse(
{'message': "Internal server error - model configuation incompatible with API (code 10003)"},
status=500
)
# a project must be specified in any request to a model of this type
if 'project__id' not in request.GET and 'project' not in request.GET:
return JsonResponse({'message': "Query not complete - Project must be specified"}, status=400)
if request.user.groups.filter(name='%s_superusers' % kwargs['app']).count() > 0:
return function(request, *args, **kwargs)
# Here we need to grab the user fields and add them to the query against the user
project_model = apps.get_model(kwargs['app'], 'Project')
user_fields = project_model.get_user_fields()
# first add the user as a field since this is project_or_user
query = Q(api_views.get_query_tuple('ForeignKey', 'user', request.user))
for field in user_fields:
query_tuple = api_views.get_query_tuple(user_fields[field], field, request.user)
query |= Q(('project__%s' % (query_tuple[0]), query_tuple[1]))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
elif availability == 'public_or_user':
# anyone can see it if it has a public flag set to True if not then only owner or superuser
# this is for mixed tables like transcriptions and verses
# All hybrid public models need a 'public' entry in the schema
# return server error if not
if 'public' not in target.get_fields():
return JsonResponse(
{'message': "Internal server error - model configuation incompatible with API (code 10004)"},
status=500
)
if not request.user.is_authenticated: # we are not logged in
# then you only get the public ones
# assumes a public boolean attribute on the model (which is okay because we have checked above)
query = Q(('public', True))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
if request.user.groups.filter(name='%s_superusers' % kwargs['app']).count() > 0:
return function(request, *args, **kwargs)
query = Q()
query |= Q(('public', True))
query |= Q(('user', request.user))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
elif availability == 'private':
# only the owner or a superuser can see it - working and draft transcriptions
if not request.user.is_authenticated: # we are not logged in
# You get nothing
return JsonResponse({'message': "Authentication required"}, status=401)
if request.user.groups.filter(name='%s_superusers' % kwargs['app']).count() > 0:
return function(request, *args, **kwargs)
query = Q(('user', request.user))
kwargs['supplied_filter'] = query
return function(request, *args, **kwargs)
else:
# just to be sure
return JsonResponse(
{'message': "Internal server error - model availability incompatible with API (code 10005)"},
status=500
)
return wrap