我们的项目正在使用 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