2

我们的项目正在使用 Django、Django-Rest-Framework 和 drf_yasg 大摇大摆地记录我们的 API。我创建了一个序列化程序,并将它与我的 POST 方法一起使用。一切正常,所有字段都出现在我的招摇页面上。我使用相同的序列化程序创建了一个删除方法,并且 swagger 文档仅将第一组字段显示为 API 的参数。

我试着用

@swagger_auto_schema(
    method='post',
    operation_description="POST /Rules/list/",
    responses={200: listResponseSerializer()},
)
@swagger_auto_schema(
@swagger_auto_schema(responses={200: listResponseSerializer()})
    method='delete',
    operation_description="DELETE /Rules/list/",
    responses={200: listResponseSerializer()},
)
@action(detail=False, methods=['post', 'delete'])

并有一个单一的方法。文档页面中立即缺少 post 和 delete 方法。

以及使用时 @swagger_auto_schema(responses={200: listResponseSerializer()})

Post 方法有效,但 delete 方法无效。

此捕获显示了 POST 方法的良好结果,其中所有字段都显示在 API 文档中。过滤器、参数等。

在此处输入图像描述

使用 DELETE,它只显示过滤器并忽略序列化程序中存在的所有其他字段。

在此处输入图像描述

希望有人可以帮助我解决这个问题。我现在正在为此苦苦挣扎几天。我不明白我错过了什么,或者我不理解 django 和 swagger 之间的哪些集成,需要帮助。

视图.py:

    class ActionAPIView(generics.GenericAPIView):
        """
        Base class for all Action views so the filtering is automatically inserted

        """
        filter_class = RESTRuleFilter

        def get_queryset(self):
            return Rule.objects.allowed_to_user(self.request.user)

        def filter_queryset(self, queryset):
            serializer = self.get_serializer(data=self.request.data)
            if serializer.is_valid():
                if 'filter' in serializer.validated_data:
                    filter = RESTRuleFilter(serializer.validated_data['filter'], queryset=queryset)
                    return filter.qs

            return queryset

        def get_serializer(self, *args, **kwargs):
            return listSerializer(*args, **kwargs)


    class listItemsApi(ActionAPIView):
        """
        post:
            add one or more Items to selected Rule list

        delete:
            remove one or more Items to selected Rule list

        """
        def get_queryset(self):
            return Rule.objects.allowed_to_user(self.request.user)

        def get_serializer(self, *args, **kwargs):
            return listPostSerializer(*args, **kwargs)

        def get_delete_serializer(self, *args, **kwargs):
            return listSerializer(*args, **kwargs)

        def get_channel(self, list_id):
            bu_list = []
            for i in list_id:
                bu = Channel.objects.allowed_to_user(self.request.user).get(
                    pk=i
                )
                bu_list.append({"pk": bu.id, "name": bu.name})

            return bu_list



        @swagger_auto_schema(responses={200: listResponseSerializer()})
        def post(self, request, *args, **kwargs):
            queryset = self.filter_queryset(self.get_queryset())
            serializer = self.get_serializer(data=request.data)

            if serializer.is_valid():
                selected_Rules = serializer.get_selected_Rules(queryset)
                bu_ids, cluster_list, Rules_id = self.build_sets(selected_Rules)
                bu = self.get_channel(bu_ids)

                task_id, timestamp = "sample_task_001"


                response = {
                    "channels": bu,
                    "task_id": task_id,
                }

                return Response(response, status=status.HTTP_200_OK)
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

在 serializer.py 中:

    class listItemsSerializer(serializers.Serializer):
        block_item = serializers.CharField(required=False, allow_blank=True)
        never_log_item = serializers.BooleanField(required=False, default=False)
        never_learn_item = serializers.BooleanField(required=False, default=False)
        description = serializers.CharField(required=False, allow_blank=True)



    class listFiltersSerializer(serializers.Serializer):
        name = serializers.CharField(required=False, allow_blank=True)
        cluster = serializers.CharField(required=False, allow_blank=True)
        channel = serializers.CharField(required=False, allow_blank=True)
        partition = serializers.CharField(required=False, allow_blank=True)


    class listSerializer(serializers.Serializer):
        filters = listFiltersSerializer(required=False, help_text="list of Rules filters")
        items = serializers.ListField(child=serializers.CharField(), required=True)
        selected = serializers.ListField(child=serializers.IntegerField(), required=False, write_only=True)

        select_all = serializers.BooleanField(
            default=False,
            write_only=True,
            help_text=("Set to True to select all objects"),
        )

        def validate(self, data):
            """
            Check if either both selected_status and select_all are configured or 
            we have a list of selected entries in selected
            """
            user = get_current_user()
            if data["select_all"]:
                return data

            if data["select_all"] is False and (
                "selected" not in data or not data["selected"]
            ):
                raise serializers.ValidationError(
                    "You must specify either select_all or selected"
                )

            found_count = 0
            if "selected" in data and data["selected"]:
                for d in data["selected"]:
                    if Rule.objects.allowed_to_user(user).filter(pk=d).exists():
                        found_count += 1

                if found_count != len(data["selected"]):
                    raise serializers.ValidationError(
                        "Some of the selected items are invalid"
                    )

            return data

        def get_selected_Rules(self, queryset=None):
            if queryset is None:
                queryset = Rule.objects.all()

            if self.validated_data.get('select_all', False):
                return queryset
            else:
                return queryset.filter(pk__in=self.validated_data['selected'])


    class listPostSerializer(listSerializer):
        params = listItemsSerializer(
            required=True, help_text="list of Itemss and setting to apply on selected Rules"
        )

过滤器.py:

    class RESTRuleFilter(RESTFilterSet):
        name = django_filters.CharFilter(
            lookup_expr="icontains", help_text="Rule name containing"
        )
        cluster = django_filters.ModelMultipleChoiceFilter(
            queryset=Cluster.objects.all()
        )
        channel = django_filters.ModelMultipleChoiceFilter(
            queryset=Channel.objects.all(), method="subfilter"
        )
        partition = django_filters.CharFilter(
            lookup_expr="icontains", help_text="Partition name containing"
        )
        enforcement_mode = django_filters.CharFilter(
            lookup_expr="icontains",
            label="Enforcement Mode",
            help_text="Enforcement Mode  containing",
        )
        signatureset_name = django_filters.CharFilter(
            label="Rule containing a signature set matching this string",
            field_name="signatureset",
            method="subfilter",
        )

        class Meta:
            model = Rule
            fields = (
                "name",
                "cluster",
                "channel",
                "partition",
                "xxxxxxx",
                "yyyyyyy",
            )

网址.py:

    schema_view = get_schema_view(
        openapi.Info(
            title="WAF Management Portal API",
            default_version="v1",
            description="REST api for interaction between Frontend and Backend.",
            contact=openapi.Contact(email="xxx@email.com"),
        ),
        public=True,
        permission_classes=(permissions.AllowAny,),
    )

    urlpatterns = [
        path(
            "",
            schema_view.with_ui("swagger", cache_timeout=0),
            name="schema-swagger-ui",
        ),
        path(
            "rule/", ruleListAPIView.as_view(), name="rule_list_view_api"
        ),
        path(
            "rules/list/", listItemsApi.as_view(), name="list_api"
        ),

    ]

模型.py:

    class Channel(models.Model):
        name = models.CharField(max_length=1024, db_index=True, unique=True)

    class Cluster(models.Model):
        name = models.CharField(max_length=1024, db_index=True, unique=True)
        channel = models.ForeignKey(
            Channel, on_delete=models.SET_NULL, null=True
        )

    class rule(models.Model):
        app_id = models.CharField(max_length=32, db_index=False)
        cluster = models.ForeignKey(
            Cluster, on_delete=models.CASCADE, related_name="rules", null=True
        )
        name = models.CharField(max_length=1024, db_index=True)
        partition = models.CharField(max_length=128, db_index=True)

        objects = ruleManager()

        class Meta:
            verbose_name_plural = "rules"
            unique_together = ("app_id", "cluster")
            indexes = [models.Index(fields=["app_id", "cluster"])]

        def __str__(self):
            return self.full_path
4

0 回答 0