DRF Swagger自定义的action文档参数实现

from django import forms


class RegisterForm(forms.Form):
    name = forms.CharField(label="name", required=True)

@api_view(["GET", "POST"])
def register(request):
    if request.method == 'GET':
        form = RegisterForm(request.GET)
    else:
        form = RegisterForm(request.data)
    
    if not form.is_valid():
        raise_as_form(form)

    ip = request.META["REMOTE_HOST"]
    REGISTER_NODE_CACHE[ip] = form.data["name"]
    node = Node.objects.filter(ip=ip).first()

    if not node:
        raise ParseError("%s 没有创建节点" % ip)
    
    jobs = Job.objects.filter(src=node)

    serializer = JobSerializer(jobs, many=True)

    return Response(serializer.data)

# -*- coding: utf-8 -*-
import six, yaml

if six.PY3:
    from urllib.parse import urljoin
else:
    from urlparse import urljoin

from rest_framework.compat import coreapi
from rest_framework.schemas import SchemaGenerator
from rest_framework_swagger import renderers
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework import exceptions
from rest_framework.permissions import AllowAny
from rest_framework.schemas.generators import is_custom_action
from rest_framework.schemas.inspectors import AutoSchema


class CustomViewSchema(AutoSchema):

    def get_link(self, path, method, base_url):
        """
        __doc__ in yaml format, eg:
        desc: the desc of this api.
        parameters:
        - name: mobile
          desc: the mobile number
          type: string
          required: true
          location: query
        - name: promotion
          desc: the activity id
          type: int
          required: true
          location: form
        """
        if hasattr(self.view, 'action'):
            # Viewsets have explicitly named actions.
            action = self.view.action
        else:
            action = ''

        if not is_custom_action(action):
            # print(path, 'is api view')
            return super(CustomViewSchema, self).get_link(path, method, base_url)
            
        fields = self.get_path_fields(path, method)

        yaml_doc = None
        if self.view and self.view.__doc__:
            try:
                yaml_doc = yaml.load(self.view.__doc__)
            except:
                yaml_doc = None
        
        # print yaml_doc
        if yaml_doc and 'desc' in yaml_doc:
            desc = yaml_doc.get('desc', '')
            _method_desc = desc
            params = yaml_doc.get('parameters', [])
            for i in params:
                _name = i.get('name')
                _desc = i.get('desc')
                _required = i.get('required', True)
                _type = i.get('type', 'string')
                _location = i.get('location', 'query')
                f = coreapi.Field(
                    name=_name,
                    location=_location,
                    required=_required,
                    description=_desc,
                    type=_type
                )
                fields.append(f)
        else:
            _method_desc = self.view.__doc__ if self.view and self.view.__doc__ else ''
            fields += self.get_serializer_fields(path, method)
        fields += self.get_pagination_fields(path, method)
        fields += self.get_filter_fields(path, method)

        if fields and any([field.location in ('form', 'body') for field in fields]):
            encoding = self.get_encoding(path, method)
        else:
            encoding = None

        if base_url and path.startswith('/'):
            path = path[1:]

        return coreapi.Link(
            url=urljoin(base_url, path),
            action=method.lower(),
            encoding=encoding,
            fields=fields,
            description=_method_desc
        )

@api_view(["GET", "POST"])
@schema(CustomViewSchema())
def register(request):
    """
    desc: 注册 agent 接口
    parameters:
    - name: name
      desc: The agent host name
      type: string
      required: true
      location: form
    """
    if request.method == 'GET':
        form = RegisterForm(request.GET)
    else:
        form = RegisterForm(request.data)
    
    if not form.is_valid():
        raise_as_form(form)

    ip = request.META["REMOTE_HOST"]
    REGISTER_NODE_CACHE[ip] = form.data["name"]
    node = Node.objects.filter(ip=ip).first()

    if not node:
        raise ParseError("%s 没有创建节点" % ip)
    
    jobs = Job.objects.filter(src=node)

    serializer = JobSerializer(jobs, many=True)

    return Response(serializer.data)

class XXXXXXViewSet(viewsets.ModelViewSet):
    schema = CustomViewSchema()

    @action(methods=['post'], detail=True, url_path='set-xy')
    def set_xy(self, request, pk=None):
        """
        desc: Set Node x y position.
        parameters:
        - name: x
          desc: The position of x.
          type: float
          required: true
          location: form
        - name: y
          desc: The position of y.
          type: float
          required: true
          location: form
        """
        obj = self.get_object()
        form = SetXYForm(request.data)

        if not form.is_valid():
            raise_as_form(form)

        ...