<s> from django . db . models import Q <EOL> from django . http import Http404 <EOL> from django . shortcuts import get_object_or_404 <EOL> from django . utils import six <EOL> from django . utils . translation import ugettext as _ <EOL> from django . core . exceptions import PermissionDenied <EOL> from rest_framework import status <EOL> from rest_framework . decorators import detail_route <EOL> from rest_framework . response import Response <EOL> from rest_framework . viewsets import ModelViewSet <EOL> from rest_framework . exceptions import ParseError <EOL> from rest_framework . settings import api_settings <EOL> from onadata . apps . api . viewsets . xform_viewset import custom_response_handler <EOL> from onadata . apps . api . tools import add_tags_to_instance <EOL> from onadata . apps . logger . models . xform import XForm <EOL> from onadata . apps . logger . models . instance import Instance <EOL> from onadata . libs . renderers import renderers <EOL> from onadata . libs . mixins . anonymous_user_public_forms_mixin import ( <EOL> AnonymousUserPublicFormsMixin ) <EOL> from onadata . apps . api . permissions import XFormPermissions <EOL> from onadata . libs . serializers . data_serializer import ( <EOL> DataSerializer , DataListSerializer , DataInstanceSerializer ) <EOL> from onadata . libs import filters <EOL> from onadata . libs . utils . viewer_tools import ( <EOL> EnketoError , <EOL> get_enketo_edit_url ) <EOL> SAFE_METHODS = [ '<STR_LIT:GET>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class DataViewSet ( AnonymousUserPublicFormsMixin , ModelViewSet ) : <EOL> """<STR_LIT>""" <EOL> renderer_classes = api_settings . DEFAULT_RENDERER_CLASSES + [ <EOL> renderers . XLSRenderer , <EOL> renderers . XLSXRenderer , <EOL> renderers . CSVRenderer , <EOL> renderers . CSVZIPRenderer , <EOL> renderers . SAVZIPRenderer , <EOL> renderers . SurveyRenderer <EOL> ] <EOL> filter_backends = ( filters . AnonDjangoObjectPermissionFilter , <EOL> filters . XFormOwnerFilter ) <EOL> permission_classes = ( XFormPermissions , ) <EOL> lookup_field = '<STR_LIT>' <EOL> lookup_fields = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> extra_lookup_fields = None <EOL> public_data_endpoint = '<STR_LIT>' <EOL> queryset = XForm . objects . all ( ) <EOL> def get_serializer_class ( self ) : <EOL> pk_lookup , dataid_lookup = self . lookup_fields <EOL> pk = self . kwargs . get ( pk_lookup ) <EOL> dataid = self . kwargs . get ( dataid_lookup ) <EOL> if pk is not None and dataid is None and pk != self . public_data_endpoint : <EOL> serializer_class = DataListSerializer <EOL> elif pk is not None and dataid is not None : <EOL> serializer_class = DataInstanceSerializer <EOL> else : <EOL> serializer_class = DataSerializer <EOL> return serializer_class <EOL> def get_object ( self ) : <EOL> obj = super ( DataViewSet , self ) . get_object ( ) <EOL> pk_lookup , dataid_lookup = self . lookup_fields <EOL> pk = self . kwargs . get ( pk_lookup ) <EOL> dataid = self . kwargs . get ( dataid_lookup ) <EOL> if pk is not None and dataid is not None : <EOL> try : <EOL> int ( dataid ) <EOL> except ValueError : <EOL> raise ParseError ( _ ( u"<STR_LIT>" <EOL> % { '<STR_LIT>' : dataid } ) ) <EOL> obj = get_object_or_404 ( Instance , pk = dataid , xform__pk = pk ) <EOL> return obj <EOL> def _get_public_forms_queryset ( self ) : <EOL> return XForm . objects . filter ( Q ( shared = True ) | Q ( shared_data = True ) ) <EOL> def _filtered_or_shared_qs ( self , qs , pk ) : <EOL> filter_kwargs = { self . lookup_field : pk } <EOL> qs = qs . filter ( ** filter_kwargs ) <EOL> if not qs : <EOL> filter_kwargs [ '<STR_LIT>' ] = True <EOL> qs = XForm . objects . filter ( ** filter_kwargs ) <EOL> if not qs : <EOL> raise Http404 ( _ ( u"<STR_LIT>" ) ) <EOL> return qs <EOL> def filter_queryset ( self , queryset , view = None ) : <EOL> qs = super ( DataViewSet , self ) . filter_queryset ( queryset ) <EOL> pk = self . kwargs . get ( self . lookup_field ) <EOL> tags = self . request . query_params . get ( '<STR_LIT>' , None ) <EOL> if tags and isinstance ( tags , six . string_types ) : <EOL> tags = tags . split ( '<STR_LIT:U+002C>' ) <EOL> qs = qs . filter ( tags__name__in = tags ) . distinct ( ) <EOL> if pk : <EOL> try : <EOL> int ( pk ) <EOL> except ValueError : <EOL> if pk == self . public_data_endpoint : <EOL> qs = self . _get_public_forms_queryset ( ) <EOL> else : <EOL> raise ParseError ( _ ( u"<STR_LIT>" % { '<STR_LIT>' : pk } ) ) <EOL> else : <EOL> qs = self . _filtered_or_shared_qs ( qs , pk ) <EOL> return qs <EOL> @ detail_route ( methods = [ '<STR_LIT:GET>' , '<STR_LIT:POST>' , '<STR_LIT>' ] , extra_lookup_fields = [ '<STR_LIT:label>' , ] ) <EOL> def labels ( self , request , * args , ** kwargs ) : <EOL> http_status = status . HTTP_400_BAD_REQUEST <EOL> instance = self . get_object ( ) <EOL> if request . method == '<STR_LIT:POST>' : <EOL> if add_tags_to_instance ( request , instance ) : <EOL> http_status = status . HTTP_201_CREATED <EOL> tags = instance . tags <EOL> label = kwargs . get ( '<STR_LIT:label>' , None ) <EOL> if request . method == '<STR_LIT:GET>' and label : <EOL> data = [ tag [ '<STR_LIT:name>' ] for tag in <EOL> tags . filter ( name = label ) . values ( '<STR_LIT:name>' ) ] <EOL> elif request . method == '<STR_LIT>' and label : <EOL> count = tags . count ( ) <EOL> tags . remove ( label ) <EOL> http_status = status . HTTP_200_OK if count == tags . count ( ) else status . HTTP_404_NOT_FOUND <EOL> data = list ( tags . names ( ) ) <EOL> else : <EOL> data = list ( tags . names ( ) ) <EOL> if request . method == '<STR_LIT:GET>' : <EOL> http_status = status . HTTP_200_OK <EOL> return Response ( data , status = http_status ) <EOL> @ detail_route ( methods = [ '<STR_LIT:GET>' ] ) <EOL> def enketo ( self , request , * args , ** kwargs ) : <EOL> self . object = self . get_object ( ) <EOL> data = { } <EOL> if isinstance ( self . object , XForm ) : <EOL> raise ParseError ( _ ( u"<STR_LIT>" ) ) <EOL> elif ( isinstance ( self . object , Instance ) ) : <EOL> if request . user . has_perm ( "<STR_LIT>" , self . object . xform ) : <EOL> return_url = request . query_params . get ( '<STR_LIT>' ) <EOL> if not return_url : <EOL> raise ParseError ( _ ( u"<STR_LIT>" ) ) <EOL> try : <EOL> data [ "<STR_LIT:url>" ] = get_enketo_edit_url ( <EOL> request , self . object , return_url ) <EOL> except EnketoError as e : <EOL> data [ '<STR_LIT>' ] = "<STR_LIT:{}>" . format ( e ) <EOL> else : <EOL> raise PermissionDenied ( _ ( u"<STR_LIT>" ) ) <EOL> return Response ( data = data ) <EOL> def destroy ( self , request , * args , ** kwargs ) : <EOL> self . object = self . get_object ( ) <EOL> if isinstance ( self . object , XForm ) : <EOL> raise ParseError ( _ ( u"<STR_LIT>" ) ) <EOL> elif isinstance ( self . object , Instance ) : <EOL> if request . user . has_perm ( "<STR_LIT>" , self . object . xform ) : <EOL> self . object . delete ( ) <EOL> else : <EOL> raise PermissionDenied ( _ ( u"<STR_LIT>" <EOL> u"<STR_LIT>" ) ) <EOL> return Response ( status = status . HTTP_204_NO_CONTENT ) <EOL> def retrieve ( self , request , * args , ** kwargs ) : <EOL> data_id = str ( kwargs . get ( '<STR_LIT>' ) ) <EOL> _format = kwargs . get ( '<STR_LIT>' ) <EOL> if not data_id . isdigit ( ) : <EOL> raise ParseError ( _ ( u"<STR_LIT>" ) ) <EOL> try : <EOL> instance = Instance . objects . get ( pk = data_id ) <EOL> if _format == '<STR_LIT>' or _format is None : <EOL> return Response ( instance . json ) <EOL> elif _format == '<STR_LIT>' : <EOL> return Response ( instance . xml ) <EOL> else : <EOL> raise ParseError ( <EOL> _ ( u"<STR_LIT>" % <EOL> { '<STR_LIT>' : _format } ) <EOL> ) <EOL> except Instance . DoesNotExist : <EOL> raise ParseError ( <EOL> _ ( u"<STR_LIT>" % <EOL> { '<STR_LIT>' : data_id } ) <EOL> ) <EOL> def list ( self , request , * args , ** kwargs ) : <EOL> lookup_field = self . lookup_field <EOL> lookup = self . kwargs . get ( lookup_field ) <EOL> if lookup_field not in kwargs . keys ( ) : <EOL> self . object_list = self . filter_queryset ( self . get_queryset ( ) ) <EOL> serializer = self . get_serializer ( self . object_list , many = True ) <EOL> return Response ( serializer . data ) <EOL> if lookup == self . public_data_endpoint : <EOL> self . object_list = self . _get_public_forms_queryset ( ) <EOL> page = self . paginate_queryset ( self . object_list ) <EOL> if page is not None : <EOL> serializer = self . get_pagination_serializer ( page ) <EOL> else : <EOL> serializer = self . get_serializer ( self . object_list , many = True ) <EOL> return Response ( serializer . data ) <EOL> xform = self . get_object ( ) <EOL> query = request . GET . get ( "<STR_LIT>" , { } ) <EOL> export_type = kwargs . get ( '<STR_LIT>' ) <EOL> if export_type is None or export_type in [ '<STR_LIT>' ] : <EOL> res = super ( DataViewSet , self ) . list ( request , * args , ** kwargs ) <EOL> res . data = res . data [ <NUM_LIT:0> ] <EOL> return res <EOL> return custom_response_handler ( request , xform , query , export_type ) </s>
