-
-
Notifications
You must be signed in to change notification settings - Fork 124
/
briefcase_api.py
279 lines (227 loc) · 10.2 KB
/
briefcase_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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
# coding: utf-8
from xml.dom import NotFoundErr
from django.conf import settings
from django.core.files import File
from django.core.validators import ValidationError
from django.contrib.auth.models import User
from django.http import Http404
from django.utils.translation import ugettext as _
from django.utils import six
from rest_framework import exceptions
from rest_framework import mixins
from rest_framework import status
from rest_framework import viewsets
from rest_framework import permissions
from rest_framework.generics import get_object_or_404
from rest_framework.renderers import BrowsableAPIRenderer
from rest_framework.response import Response
from rest_framework.decorators import action
from onadata.apps.api.tools import get_media_file_response
from onadata.apps.api.permissions import ViewDjangoObjectPermissions
from onadata.apps.logger.models.attachment import Attachment
from onadata.apps.logger.models.instance import Instance
from onadata.apps.logger.models.xform import XForm
from onadata.apps.main.models.meta_data import MetaData
from onadata.apps.main.models.user_profile import UserProfile
from onadata.libs import filters
from onadata.libs.authentication import DigestAuthentication
from onadata.libs.mixins.openrosa_headers_mixin import OpenRosaHeadersMixin
from onadata.libs.renderers.renderers import TemplateXMLRenderer
from onadata.libs.serializers.xform_serializer import XFormListSerializer
from onadata.libs.serializers.xform_serializer import XFormManifestSerializer
from onadata.libs.utils.logger_tools import publish_form, publish_xml_form, \
get_instance_or_404
def _extract_uuid(text):
if isinstance(text, six.string_types):
form_id_parts = text.split('/')
if form_id_parts.__len__() < 2:
raise ValidationError(_("Invalid formId %s." % text))
text = form_id_parts[1]
text = text[text.find("@key="):-1].replace("@key=", "")
if text.startswith("uuid:"):
text = text.replace("uuid:", "")
return text
def _extract_id_string(formId):
if isinstance(formId, six.string_types):
return formId[0:formId.find('[')]
return formId
def _parse_int(num):
try:
return num and int(num)
except ValueError:
pass
class DoXmlFormUpload:
def __init__(self, xml_file, user):
self.xml_file = xml_file
self.user = user
def publish(self):
return publish_xml_form(self.xml_file, self.user)
class BriefcaseApi(OpenRosaHeadersMixin, mixins.CreateModelMixin,
mixins.RetrieveModelMixin, mixins.ListModelMixin,
viewsets.GenericViewSet):
"""
Implements the [Briefcase Aggregate API](\
https://code.google.com/p/opendatakit/wiki/BriefcaseAggregateAPI).
"""
filter_backends = (filters.AnonDjangoObjectPermissionFilter,)
queryset = XForm.objects.all()
permission_classes = (permissions.IsAuthenticated,
ViewDjangoObjectPermissions)
renderer_classes = (TemplateXMLRenderer, BrowsableAPIRenderer)
serializer_class = XFormListSerializer
template_name = 'openrosa_response.xml'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Respect DEFAULT_AUTHENTICATION_CLASSES, but also ensure that the
# previously hard-coded authentication classes are included first
authentication_classes = [
DigestAuthentication,
]
self.authentication_classes = authentication_classes + [
auth_class
for auth_class in self.authentication_classes
if auth_class not in authentication_classes
]
def get_object(self):
formId = self.request.GET.get('formId', '')
id_string = _extract_id_string(formId)
uuid = _extract_uuid(formId)
username = self.kwargs.get('username')
obj = get_instance_or_404(xform__user__username__iexact=username,
xform__id_string__exact=id_string,
uuid=uuid)
self.check_object_permissions(self.request, obj.xform)
return obj
def filter_queryset(self, queryset):
username = self.kwargs.get('username')
if username is None and self.request.user.is_anonymous:
# raises a permission denied exception, forces authentication
self.permission_denied(self.request)
if username is not None and self.request.user.is_anonymous:
profile = get_object_or_404(
UserProfile, user__username=username.lower())
if profile.require_auth:
# raises a permission denied exception, forces authentication
self.permission_denied(self.request)
else:
queryset = queryset.filter(user=profile.user)
else:
queryset = super().filter_queryset(queryset)
formId = self.request.GET.get('formId', '')
if formId.find('[') != -1:
formId = _extract_id_string(formId)
xform = get_object_or_404(queryset, id_string__exact=formId)
self.check_object_permissions(self.request, xform)
instances = Instance.objects.filter(xform=xform).order_by('pk')
num_entries = self.request.GET.get('numEntries')
cursor = self.request.GET.get('cursor')
cursor = _parse_int(cursor)
if cursor:
instances = instances.filter(pk__gt=cursor)
num_entries = _parse_int(num_entries)
if num_entries:
instances = instances[:num_entries]
if instances.count():
last_instance = instances[instances.count() - 1]
self.resumptionCursor = last_instance.pk
elif instances.count() == 0 and cursor:
self.resumptionCursor = cursor
else:
self.resumptionCursor = 0
return instances
def create(self, request, *args, **kwargs):
if request.method.upper() == 'HEAD':
return Response(status=status.HTTP_204_NO_CONTENT,
headers=self.get_openrosa_headers(request),
template_name=self.template_name)
xform_def = request.FILES.get('form_def_file', None)
response_status = status.HTTP_201_CREATED
username = kwargs.get('username')
form_user = (username and get_object_or_404(User, username=username)) \
or request.user
if not request.user.has_perm(
'can_add_xform',
UserProfile.objects.get_or_create(user=form_user)[0]
):
raise exceptions.PermissionDenied(
detail=_("User %(user)s has no permission to add xforms to "
"account %(account)s" %
{'user': request.user.username,
'account': form_user.username}))
data = {}
if isinstance(xform_def, File):
do_form_upload = DoXmlFormUpload(xform_def, form_user)
dd = publish_form(do_form_upload.publish)
if isinstance(dd, XForm):
data['message'] = _(
"%s successfully published." % dd.id_string)
else:
data['message'] = dd['text']
response_status = status.HTTP_400_BAD_REQUEST
else:
data['message'] = _("Missing xml file.")
response_status = status.HTTP_400_BAD_REQUEST
return Response(data, status=response_status,
headers=self.get_openrosa_headers(request,
location=False),
template_name=self.template_name)
def list(self, request, *args, **kwargs):
self.object_list = self.filter_queryset(self.get_queryset())
data = {'instances': self.object_list,
'resumptionCursor': self.resumptionCursor}
return Response(data,
headers=self.get_openrosa_headers(request,
location=False),
template_name='submissionList.xml')
def retrieve(self, request, *args, **kwargs):
self.object = self.get_object()
submission_xml_root_node = self.object.get_root_node()
submission_xml_root_node.setAttribute(
'instanceID', 'uuid:%s' % self.object.uuid)
submission_xml_root_node.setAttribute(
'submissionDate', self.object.date_created.isoformat()
)
# Added this because of https://github.com/onaio/onadata/pull/2139
# Should bring support to ODK v1.17+
if settings.SUPPORT_BRIEFCASE_SUBMISSION_DATE:
# Remove namespace attribute if any
try:
submission_xml_root_node.removeAttribute('xmlns')
except NotFoundErr:
pass
data = {
'submission_data': submission_xml_root_node.toxml(),
'media_files': Attachment.objects.filter(instance=self.object),
'host': request.build_absolute_uri().replace(
request.get_full_path(), '')
}
return Response(
data,
headers=self.get_openrosa_headers(request, location=False),
template_name='downloadSubmission.xml',
)
@action(detail=True, methods=['GET'])
def manifest(self, request, *args, **kwargs):
self.object = self.get_object()
object_list = MetaData.objects.filter(
data_type__in=MetaData.MEDIA_FILES_TYPE, xform=self.object
)
context = self.get_serializer_context()
serializer = XFormManifestSerializer(object_list, many=True,
context=context)
return Response(serializer.data,
headers=self.get_openrosa_headers(request,
location=False))
@action(detail=True, methods=['GET'])
def media(self, request, *args, **kwargs):
self.object = self.get_object()
pk = kwargs.get('metadata')
if not pk:
raise Http404()
meta_obj = get_object_or_404(
MetaData,
data_type__in=MetaData.MEDIA_FILES_TYPE,
xform=self.object,
pk=pk,
)
return get_media_file_response(meta_obj, request)