111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
''' Endpoints for OSS. '''
|
|
from typing import cast
|
|
|
|
from drf_spectacular.utils import extend_schema, extend_schema_view
|
|
from rest_framework import generics
|
|
from rest_framework import status as c
|
|
from rest_framework import viewsets
|
|
from rest_framework.decorators import action
|
|
from rest_framework.request import Request
|
|
from rest_framework.response import Response
|
|
|
|
from shared import permissions
|
|
|
|
from .. import models as m
|
|
from .. import serializers as s
|
|
|
|
|
|
@extend_schema(tags=['OSS'])
|
|
@extend_schema_view()
|
|
class OssViewSet(viewsets.GenericViewSet, generics.ListAPIView, generics.RetrieveAPIView):
|
|
''' Endpoint: OperationSchema. '''
|
|
queryset = m.LibraryItem.objects.filter(item_type=m.LibraryItemType.OPERATION_SCHEMA)
|
|
serializer_class = s.LibraryItemSerializer
|
|
|
|
def _get_schema(self) -> m.OperationSchema:
|
|
return m.OperationSchema(cast(m.LibraryItem, self.get_object()))
|
|
|
|
def get_permissions(self):
|
|
''' Determine permission class. '''
|
|
if self.action in [
|
|
'operation_create',
|
|
'operation_delete'
|
|
]:
|
|
permission_list = [permissions.ItemEditor]
|
|
elif self.action in ['details']:
|
|
permission_list = [permissions.ItemAnyone]
|
|
else:
|
|
permission_list = [permissions.Anyone]
|
|
return [permission() for permission in permission_list]
|
|
|
|
@extend_schema(
|
|
summary='get operations data',
|
|
tags=['OSS'],
|
|
request=None,
|
|
responses={
|
|
c.HTTP_200_OK: s.OperationSchemaSerializer,
|
|
c.HTTP_404_NOT_FOUND: None
|
|
}
|
|
)
|
|
@action(detail=True, methods=['get'], url_path='details')
|
|
def details(self, request: Request, pk):
|
|
''' Endpoint: Detailed OSS data. '''
|
|
serializer = s.OperationSchemaSerializer(cast(m.LibraryItem, self.get_object()))
|
|
return Response(
|
|
status=c.HTTP_200_OK,
|
|
data=serializer.data
|
|
)
|
|
|
|
@extend_schema(
|
|
summary='create operation',
|
|
tags=['OSS'],
|
|
request=s.OperationCreateSerializer(),
|
|
responses={
|
|
c.HTTP_201_CREATED: s.NewOperationResponse,
|
|
c.HTTP_403_FORBIDDEN: None
|
|
}
|
|
)
|
|
@action(detail=True, methods=['post'], url_path='operation-create')
|
|
def operation_create(self, request: Request, pk):
|
|
''' Create new operation. '''
|
|
schema = self._get_schema()
|
|
serializer = s.OperationCreateSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
data = serializer.validated_data
|
|
new_operation = schema.create_operation(*data)
|
|
schema.item.refresh_from_db()
|
|
response = Response(
|
|
status=c.HTTP_201_CREATED,
|
|
data={
|
|
'new_operation': s.OperationSerializer(new_operation).data,
|
|
'schema': s.OperationSchemaSerializer(schema.item).data
|
|
}
|
|
)
|
|
return response
|
|
|
|
# @extend_schema(
|
|
# summary='delete operation',
|
|
# tags=['RSForm'],
|
|
# request=s.CstListSerializer,
|
|
# responses={
|
|
# c.HTTP_200_OK: s.RSFormParseSerializer,
|
|
# c.HTTP_403_FORBIDDEN: None,
|
|
# c.HTTP_404_NOT_FOUND: None
|
|
# }
|
|
# )
|
|
# @action(detail=True, methods=['patch'], url_path='operation-delete')
|
|
# def operation_delete(self, request: Request, pk):
|
|
# ''' Endpoint: Delete operation. '''
|
|
# schema = self._get_schema()
|
|
# serializer = s.CstListSerializer(
|
|
# data=request.data,
|
|
# context={'schema': schema.item}
|
|
# )
|
|
# serializer.is_valid(raise_exception=True)
|
|
# schema.delete_cst(serializer.validated_data['items'])
|
|
# schema.item.refresh_from_db()
|
|
# return Response(
|
|
# status=c.HTTP_200_OK,
|
|
# data=s.RSFormParseSerializer(schema.item).data
|
|
# )
|