#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
###############################################################################
# #
# Copyright (c) 2016 Idiap Research Institute, http://www.idiap.ch/ #
# Contact: beat.support@idiap.ch #
# #
# This file is part of the beat.web module of the BEAT platform. #
# #
# Commercial License Usage #
# Licensees holding valid commercial BEAT licenses may use this file in #
# accordance with the terms contained in a written agreement between you #
# and Idiap. For further information contact tto@idiap.ch #
# #
# Alternatively, this file may be used under the terms of the GNU Affero #
# Public License version 3 as published by the Free Software and appearing #
# in the file LICENSE.AGPL included in the packaging of this file. #
# The BEAT platform is distributed in the hope that it will be useful, but #
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY #
# or FITNESS FOR A PARTICULAR PURPOSE. #
# #
# You should have received a copy of the GNU Affero Public License along #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/. #
# #
###############################################################################
from django.conf import settings
from django.shortcuts import get_object_or_404
from django.utils import six
from rest_framework.response import Response
from rest_framework import permissions
from rest_framework import generics
from rest_framework import status
from rest_framework.views import APIView
from datetime import datetime
from .models import Attestation
from .serializers import AttestationSerializer
from .exceptions import AlreadyUnlockedError
from ..experiments.models import Experiment
from ..common.models import Shareable
from ..common.exceptions import ShareError
from ..common.responses import BadRequestResponse, ForbiddenResponse
import beat.core.experiment
import beat.core.toolchain
import simplejson as json
#----------------------------------------------------------
[docs]class CreateAttestationView(APIView):
"""
Create a new attestation
"""
permission_classes = [permissions.IsAuthenticated]
[docs] def post(self, request):
data = request.data
if 'experiment' not in data:
return BadRequestResponse('Missing field: experiment')
if not(isinstance(data['experiment'], six.string_types)):
return BadRequestResponse('Invalid field type: "experiment" must be a string')
declaration_id = beat.core.experiment.Storage(settings.PREFIX,
data['experiment'])
# Check that the user is the author of the experiment
if declaration_id.username != request.user.username:
return ForbiddenResponse()
# Retrieve the experiment
try:
toolchain_id = beat.core.toolchain.Storage(settings.PREFIX,
declaration_id.toolchain)
experiment = Experiment.objects.get(
author__username=declaration_id.username,
toolchain__author__username=toolchain_id.username,
toolchain__name=toolchain_id.name,
toolchain__version=toolchain_id.version,
name=declaration_id.name
)
except:
return Response(status=404)
# Check that an attestation doesn't already exists for that experiment
if experiment.has_attestation():
return BadRequestResponse('This experiment already has an attestation')
# Check that the experiment is done
if experiment.end_date is None:
return BadRequestResponse("This experiment isn't done yet")
# Retrieve and process the list of algorithms referenced by the declaration, and
# their referenced data formats
needed_formats = []
needed_algorithms = []
for algorithm in experiment.referenced_algorithms.iterator():
needed_formats.extend(algorithm.all_needed_dataformats())
needed_algorithms.append(algorithm)
needed_formats = list(set(needed_formats))
needed_algorithms = list(set(needed_algorithms))
# Process the list of referenced dataformats
other_needed_formats = filter(lambda x: x.author != experiment.author, needed_formats)
# Ensure that all needed dataformats from other users have the necessary sharing
# preferences
errors = []
for needed_format in other_needed_formats:
errors.extend(needed_format.is_accessible())
if len(errors) > 0:
return Response(errors, status=400)
# Process the list of referenced algorithms
own_needed_algorithms = filter(lambda x: x.author == experiment.author, needed_algorithms)
other_needed_algorithms = filter(lambda x: x.author != experiment.author, needed_algorithms)
# Ensure that all needed algorithms from other users have the necessary sharing
# preferences
errors = []
for needed_algorithm in other_needed_algorithms:
errors.extend(needed_algorithm.is_accessible(public=False))
if len(errors) > 0:
return Response(errors, status=400)
# Create the attestation object in the database
attestation = Attestation.objects.create_attestation(experiment)
# Add the references to the data formats
for needed_dataformat in needed_formats:
attestation.dataformats.add(needed_dataformat)
# Add the references to the algorithms
for referenced_algorithm in experiment.referenced_algorithms.iterator():
attestation.algorithms.add(referenced_algorithm)
# Send the result
result = {
'number': attestation.number,
}
response = Response(result, status=201)
return response
#----------------------------------------------------------
[docs]class UnlockAttestationView(APIView):
"""
Unlock a "locked" attestation
"""
permission_classes = [permissions.IsAuthenticated]
[docs] def post(self, request, number):
data = request.data
if 'visible_algorithms' in data:
if not(isinstance(data['visible_algorithms'], list)):
return BadRequestResponse('Invalid visible_algorithms data')
if len(data['visible_algorithms']) > 0:
visible_algorithms_names = data['visible_algorithms']
else:
visible_algorithms_names = []
else:
visible_algorithms_names = []
# Retrieve the attestation
attestation = get_object_or_404(Attestation, number=number)
# Check that the user is the author of the attestation
if attestation.experiment.author != request.user:
return ForbiddenResponse()
# Unlock the experiment
try:
attestation.unlock(visible_algorithms_names=visible_algorithms_names)
except AlreadyUnlockedError:
return BadRequestResponse('This attestation is already unlocked')
except ShareError as e:
return Response(e.errors, status=400)
return Response(status=204)
#----------------------------------------------------------
[docs]class ListUserAttestationView(generics.ListAPIView):
"""
List all attestations from a given user
"""
permission_classes = [permissions.AllowAny]
serializer_class = AttestationSerializer
[docs] def get(self, request, username):
# Retrieve all the attestations of the specified user
attestations = Attestation.objects.select_related().filter(experiment__author__username=username).order_by('-creation_date')
if username != request.user.username:
attestations = attestations.exclude(locked=True)
serializer = self.get_serializer(attestations, many=True)
return Response(serializer.data)
#----------------------------------------------------------
[docs]class DeleteAttestationView(APIView):
"""
Delete given attestation if locked
"""
permission_classes = [permissions.IsAuthenticated]
[docs] def delete(self, request, number):
attestation = get_object_or_404(Attestation, number=number)
if attestation.experiment.author != request.user or not attestation.locked:
return ForbiddenResponse()
attestation.delete()
return Response(status=status.HTTP_204_NO_CONTENT)