"""
Provides all of the class-based views for the REST API.
"""
from django.db.models import Q
from rest_framework import status
from rest_framework.settings import api_settings
from rest_framework import viewsets, exceptions, status
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import (IsAuthenticated,
IsAuthenticatedOrReadOnly)
from rest_framework.response import Response
from rest_framework.decorators import detail_route, list_route
from rest_framework.pagination import (LimitOffsetPagination,
PageNumberPagination)
from annotations.serializers import *
from annotations.tasks import get_manager
from annotations.models import (VogonUser, Repository, Appellation, RelationSet,
Relation, TemporalBounds, Text, TextCollection)
from concepts.models import Concept, Type
import uuid
[docs]class StandardResultsSetPagination(PageNumberPagination):
page_size = 100
page_size_query_param = 'page_size'
max_page_size = 1000
[docs]class AnnotationFilterMixin(object):
"""
Mixin for :class:`viewsets.ModelViewSet` that provides filtering by
:class:`.Text` and :class:`.User`\.
"""
[docs] def get_queryset(self, *args, **kwargs):
queryset = super(AnnotationFilterMixin, self).get_queryset(*args, **kwargs)
textid = self.request.query_params.get('text', None)
texturi = self.request.query_params.get('text_uri', None)
userid = self.request.query_params.get('user', None)
if textid:
queryset = queryset.filter(occursIn=int(textid))
if texturi:
queryset = queryset.filter(occursIn__uri=texturi)
if userid:
queryset = queryset.filter(createdBy__pk=userid)
elif userid is not None:
queryset = queryset.filter(createdBy__pk=self.request.user.id)
return queryset
[docs]class UserViewSet(viewsets.ModelViewSet):
queryset = VogonUser.objects.all()
serializer_class = UserSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs]class RepositoryViewSet(viewsets.ModelViewSet):
queryset = Repository.objects.all()
serializer_class = RepositorySerializer
permission_classes = (IsAuthenticated, )
[docs]class RemoteCollectionViewSet(viewsets.ViewSet):
[docs] def list(self, request, repository_pk=None):
repository = Repository.objects.get(pk=repository_pk)
manager = get_manager(repository.manager)(repository.endpoint)
return Response(manager.collections())
[docs] def retrieve(self, request, pk=None, repository_pk=None):
repository = Repository.objects.get(pk=repository_pk)
manager = get_manager(repository.manager)(repository.endpoint)
return Response(manager.collection(pk))
[docs]class RemoteResourceViewSet(viewsets.ViewSet):
[docs] def list(self, request, collection_pk=None, repository_pk=None):
repository = Repository.objects.get(pk=repository_pk)
manager = get_manager(repository.manager)(repository.endpoint)
return Response(manager.collection(collection_pk))
[docs] def retrieve(self, request, pk=None, collection_pk=None, repository_pk=None):
repository = Repository.objects.get(pk=repository_pk)
manager = get_manager(repository.manager)(repository.endpoint)
return Response(manager.resource(pk))
[docs]class AppellationViewSet(AnnotationFilterMixin, viewsets.ModelViewSet):
queryset = Appellation.objects.filter(asPredicate=False)
serializer_class = AppellationSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
# pagination_class = LimitOffsetPagination
[docs] def create(self, request, *args, **kwargs):
data = request.data
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED,
headers=headers)
[docs] def get_queryset(self, *args, **kwargs):
queryset = AnnotationFilterMixin.get_queryset(self, *args, **kwargs)
concept = self.request.query_params.get('concept', None)
text = self.request.query_params.get('text', None)
thisuser = self.request.query_params.get('thisuser', False)
if thisuser:
queryset = queryset.filter(createdBy_id=self.request.user.id)
if concept:
queryset = queryset.filter(interpretation_id=concept)
if text:
queryset = queryset.filter(occursIn_id=text)
return queryset
[docs]class PredicateViewSet(AnnotationFilterMixin, viewsets.ModelViewSet):
queryset = Appellation.objects.filter(asPredicate=True)
serializer_class = AppellationSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs]class RelationSetViewSet(viewsets.ModelViewSet):
queryset = RelationSet.objects.all()
serializer_class = RelationSetSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs] def get_queryset(self, *args, **kwargs):
queryset = super(RelationSetViewSet, self).get_queryset(*args, **kwargs)
textid = self.request.query_params.getlist('text')
userid = self.request.query_params.getlist('user')
if len(textid) > 0:
queryset = queryset.filter(occursIn__in=[int(t) for t in textid])
if len(userid) > 0:
queryset = queryset.filter(createdBy__pk__in=[int(i) for i in userid])
elif userid is not None and type(userid) is not list:
queryset = queryset.filter(createdBy__pk=self.request.user.id)
thisuser = self.request.query_params.get('thisuser', False)
if thisuser:
queryset = queryset.filter(createdBy_id=self.request.user.id)
return queryset
[docs]class RelationViewSet(viewsets.ModelViewSet):
queryset = Relation.objects.all()
serializer_class = RelationSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs] def get_queryset(self, *args, **kwargs):
"""
Supports filtering by :class:`.Text`\, :class:`.User`\, node concept
type, and predicate concept type.
"""
queryset = super(RelationViewSet, self).get_queryset(*args, **kwargs)
textid = self.request.query_params.getlist('text')
userid = self.request.query_params.getlist('user')
typeid = self.request.query_params.getlist('type')
conceptid = self.request.query_params.getlist('concept')
related_concepts = self.request.query_params.getlist('related_concepts')
# Refers to the predicate's interpretation, not the predicate itself.
predicate_conceptid = self.request.query_params.getlist('predicate')
# TODO: clean this up.
if len(textid) > 0:
queryset = queryset.filter(occursIn__in=[int(t) for t in textid])
if len(typeid) > 0:
queryset = queryset.filter(source__interpretation__typed__pk__in=[int(t) for t in typeid]).filter(object__interpretation__typed__pk__in=[int(t) for t in typeid])
if len(predicate_conceptid) > 0:
queryset = queryset.filter(predicate__interpretation__pk__in=[int(t) for t in predicate_conceptid])
if len(conceptid) > 0: # Source or target concept in `concept`.
queryset = queryset.filter(Q(source__interpretation__id__in=[int(c) for c in conceptid]) | Q(object__interpretation__id__in=[int(c) for c in conceptid]))
if len(related_concepts) > 0: # Source or target concept in `concept`.
queryset = queryset.filter(Q(source__interpretation__id__in=[int(c) for c in related_concepts]) & Q(object__interpretation__id__in=[int(c) for c in related_concepts]))
if len(userid) > 0:
queryset = queryset.filter(createdBy__pk__in=[int(i) for i in userid])
elif userid is not None and type(userid) is not list:
queryset = queryset.filter(createdBy__pk=self.request.user.id)
thisuser = self.request.query_params.get('thisuser', False)
if thisuser:
queryset = queryset.filter(createdBy_id=self.request.user.id)
return queryset
# TODO: do we need this anymore?
[docs]class TemporalBoundsViewSet(viewsets.ModelViewSet, AnnotationFilterMixin):
queryset = TemporalBounds.objects.all()
serializer_class = TemporalBoundsSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs]class TextViewSet(viewsets.ModelViewSet):
queryset = Text.objects.all()
serializer_class = TextSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
# pagination_class = StandardResultsSetPagination
[docs] def get_queryset(self, *args, **kwargs):
"""
A user can see only their own :class:`.TextCollection`\s.
"""
queryset = super(TextViewSet, self).get_queryset(*args, **kwargs)
textcollectionid = self.request.query_params.get('textcollection', None)
conceptid = self.request.query_params.getlist('concept')
related_concepts = self.request.query_params.getlist('related_concepts')
uri = self.request.query_params.get('uri', None)
if textcollectionid:
queryset = queryset.filter(partOf=int(textcollectionid))
if uri:
queryset = queryset.filter(uri=uri)
if len(conceptid) > 0:
queryset = queryset.filter(appellation__interpretation__pk__in=[int(c) for c in conceptid])
if len(related_concepts) > 1:
queryset = queryset.filter(appellation__interpretation_id=int(related_concepts[0])).filter(appellation__interpretation_id=int(related_concepts[1]))
return queryset.distinct()
[docs]class TextCollectionViewSet(viewsets.ModelViewSet):
queryset = TextCollection.objects.all()
serializer_class = TextCollectionSerializer
permission_classes = (IsAuthenticated, )
[docs] def get_queryset(self, *args, **kwargs):
"""
"""
queryset = super(TextCollectionViewSet, self).get_queryset(*args, **kwargs)
userid = self.request.query_params.get('user', None)
if userid:
queryset = queryset.filter(ownedBy__pk=userid)
else:
queryset = queryset.filter(Q(ownedBy__pk=self.request.user.id) | Q(participants=self.request.user.id))
return queryset
[docs] def create(self, request, *args, **kwargs):
data = request.data
if 'ownedBy' not in data:
data['ownedBy'] = request.user.id
if 'participants' not in data:
data['participants'] = []
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(data)
return Response(serializer.data,
status=status.HTTP_201_CREATED,
headers=headers)
[docs]class TypeViewSet(viewsets.ModelViewSet):
queryset = Type.objects.all()
serializer_class = TypeSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs]class ConceptViewSet(viewsets.ModelViewSet):
queryset = Concept.objects.filter(~Q(concept_state=Concept.REJECTED))
serializer_class = ConceptSerializer
permission_classes = (IsAuthenticatedOrReadOnly, )
[docs] def create(self, request, *args, **kwargs):
data = request.data
if data['uri'] == 'generate':
data['uri'] = 'http://vogonweb.net/{0}'.format(uuid.uuid4())
if 'lemma' not in data:
data['lemma'] = data['label'].lower()
serializer = self.get_serializer(data=data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(data)
return Response(serializer.data,
status=status.HTTP_201_CREATED,
headers=headers)
[docs] def get_queryset(self, *args, **kwargs):
"""
Filter by part of speach (``pos``).
"""
queryset = super(ConceptViewSet, self).get_queryset(*args, **kwargs)
# Limit results to those with ``pos``.
pos = self.request.query_params.get('pos', None)
if pos:
if pos != 'all':
queryset = queryset.filter(pos__in=[pos.upper(), pos.lower()])
# Search Concept labels for ``search`` param.
query = self.request.query_params.get('search', None)
remote = self.request.query_params.get('remote', False)
uri = self.request.query_params.get('uri', None)
type_id = self.request.query_params.get('typed', None)
type_strict = self.request.query_params.get('strict', None)
type_uri = self.request.query_params.get('type_uri', None)
max_results = self.request.query_params.get('max', None)
if uri:
queryset = queryset.filter(uri=uri)
if type_uri:
queryset = queryset.filter(type__uri=uri)
if type_id:
if type_strict:
queryset = queryset.filter(typed_id=type_id)
else:
queryset = queryset.filter(Q(typed_id=type_id) | Q(typed=None))
if query:
if pos == 'all':
pos = None
if remote: # Spawn asynchronous calls to authority services.
search_concept.delay(query, pos=pos)
queryset = queryset.filter(label__icontains=query)
if max_results:
return queryset[:max_results]
return queryset
