Python rest_framework.views 模块，Response() 实例源码

def get(self, request, *args, **kwargs):
        apis = [{"path": "/" + path} for path in self.get_resources()]
        return Response({
            "apiVersion": rfs.SWAGGER_SETTINGS.get("api_version", ""),
            "swaggerVersion": "1.2",
            "basePath": self.get_base_path(),
            "apis": apis,
            "info": rfs.SWAGGER_SETTINGS.get("info", {
                "contact": "",
                "description": "",
                "license": "",
                "licenseUrl": "",
                "termsOfServiceUrl": "",
                "title": "",
            }),
        })

def delete(self, request, *args, **kwargs):
        if models.DB.objects.filter(id=int(kwargs['pk'])).exists():
            models.DB.objects.get(id=int(kwargs['pk'])).delete()
            return Response({'info': '????'}, status=status.HTTP_201_CREATED)
        else:
            return Response({'info': '???????'}, status=status.HTTP_406_NOT_ACCEPTABLE)

def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        db = models.DB.objects.get(id=int(kwargs['pk']))
        serializer.instance.db = db
        serializer.save()
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

def delete(self, request, *args, **kwargs):
        if models.DB.objects.get(id=int(kwargs['pk'])).dbuser.filter(user=request.data['user'],ip=request.data['ip']).exists():
            models.DB.objects.get(id=int(kwargs['pk'])).dbuser.filter(user=request.data['user'], ip=request.data['ip']).delete()
            return Response({'info': '????'}, status=status.HTTP_201_CREATED)
        else:
            return Response({'info': '?????????'}, status=status.HTTP_406_NOT_ACCEPTABLE)

def delete(self, request, *args, **kwargs):
        if models.ExtendUser.objects.filter(id=int(kwargs['pk'])).exists():
            models.ExtendUser.objects.get(id=int(kwargs['pk'])).delete()
            return Response({'info': '????'}, status=status.HTTP_201_CREATED)
        else:
            return Response({'info': '??????'}, status=status.HTTP_406_NOT_ACCEPTABLE)

def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        if models.Script.objects.get(id=int(kwargs['pk'])).scriptargs.filter(args_name=request.data['args_name']).exists():
            return Response({'info':'???????????'},status=status.HTTP_406_NOT_ACCEPTABLE)
        else:
            serializer.save()
        script = models.Script.objects.get(id=int(kwargs['pk']))
        serializer.instance.script=script
        serializer.save()
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        playbook = models.PlayBook.objects.get(id=int(kwargs['pk']))
        serializer.instance.playbook=playbook
        serializer.instance.sort = playbook.sort
        playbook.sort=playbook.sort+1
        playbook.save()
        serializer.save()
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

def delete(self, request, *args, **kwargs):
    # return self.destroy(self,request,args,kwargs)
        if models.PlayBook.objects.get(id=int(kwargs['pk'])).tasks.filter(sort=request.data['sort']).exists():
            models.PlayBook.objects.get(id=int(kwargs['pk'])).tasks.filter(sort=request.data['sort']).delete()
            return Response({'info': '????'}, status=status.HTTP_201_CREATED)
        else:
            return Response({'info': '??????????'}, status=status.HTTP_406_NOT_ACCEPTABLE)

def update(self, request, *args, **kwargs):
        tasks = models.PlayBook.objects.get(id=int(kwargs['pk'])).tasks
        taskA = tasks.get(sort=int(request.data['parta']))
        taskB = tasks.get(sort=int(request.data['partb']))
        tempsort=taskA.sort
        taskA.sort=taskB.sort
        taskB.sort=tempsort
        taskA.save()
        taskB.save()
        return Response({'info': '????'}, status=status.HTTP_201_CREATED)

def delete(self, request, *args, **kwargs):
        group = models.Group.objects.get(id=int(kwargs['pk']))
        if group.hosts.count() != 0:
            return Response({'detail': '?????????????'}, status=status.HTTP_406_NOT_ACCEPTABLE)
        else:
            group.delete()
            return Response({'detail': '????'}, status=status.HTTP_201_CREATED)

def delete(self, request, *args, **kwargs):
        host = models.Host.objects.get(id=int(kwargs['pk']))
        if host.storages.count() != 0:
            return Response({'detail': '????????????'}, status=status.HTTP_406_NOT_ACCEPTABLE)
        elif len(host.application_get()) !=0:
            return Response({'detail': '????????????'}, status=status.HTTP_406_NOT_ACCEPTABLE)
        else:
            models.Host.objects.get(id=int(kwargs['pk'])).delete()
            return Response({'detail': '????'}, status=status.HTTP_201_CREATED)

def login_user(request):
    data = json.loads(str(request.data))
    username = data['username']
    password = data['password']

    user = authenticate(username=username, password=password)

    if user is not None:
        token = Token.objects.get_or_create(user=user)

        login(request, user)
        user = UserSerializer(user)

        data = {
            'message': 'Valid User',
            'token': str(token[0]),
            'user': user.data,
            'code': 100,
        }

    else:
        data = {
            'message': 'Invalid Details',
            'code': 101
        }
    return Response(status=200, data=data)

def get(self, request, path, *args, **kwargs):
        result = Response({
            "apiVersion": rfs.SWAGGER_SETTINGS.get("api_version", ""),
            "swaggerVersion": "1.2",
            "basePath": self.api_full_uri.rstrip("/"),
            "resourcePath": "/" + path,
            "apis": NotificationsDocs.generate_notifications_docs(path),
            "models": NotificationsDocs.generate_models_docs(),
        })

        return result
# -- CODE TAKEN FROM DJANGO REST SWAGGER & MODIFIED

def post(self, request, *args, **kwargs):
        data = request.data
        serializer = UserLoginSerializer(data=data)
        if serializer.is_valid(raise_exception=True):
            user = User.objects.get(username=serializer.data['username'])
            token = self.__create_token(user=user)
            ret_json = {
                "Token": token.key,
                "error_code": 0
            }
            # new_data = serializer.data
            return Response(ret_json, status=status.HTTP_200_OK)
        return Response(serializer.error_messages, status=status.HTTP_401_UNAUTHORIZED)

def get(self, request):
        user = request.user
        user = Token.objects.get(user=user)
        user.delete()
        ret_json = {
            'error_code': 0
        }
        return Response(ret_json, status=status.HTTP_200_OK)

def post(self, request):
        request.user.notifications.mark_all_as_read()
        return Response(status=status.HTTP_200_OK)

def post(self, request, pk):
        try:
            request.user.notifications.get(pk=pk).mark_as_read()
            return Response(status=status.HTTP_200_OK)
        except Exception as e:
            print(repr(e))
            return Response(status=status.HTTP_400_BAD_REQUEST)

def get(self, request):
        """returns """
        name = self.request.GET.get('name', None)
        return Response(self.get_matching_items(name))