# -*- coding: utf-8 -*-
|
"""
|
@File : api.py
|
@Time : 2023/2/1 16:57
|
@Author : geekbing
|
@LastEditTime : -
|
@LastEditors : -
|
@Description : API视图
|
"""
|
from enum import IntEnum
|
from typing import List
|
|
from ast import literal_eval
|
|
from django.core.exceptions import ObjectDoesNotExist
|
from django.db import DataError
|
from django.db.models import Q
|
from django.utils.decorators import method_decorator
|
from django.utils import timezone
|
from drf_yasg.utils import swagger_auto_schema
|
from drf_yasg import openapi
|
from rest_framework import status
|
from rest_framework.viewsets import GenericViewSet
|
from rest_framework.response import Response
|
|
from apps.exceptions.error import RelationNotFound
|
from lunarlink import models, serializers
|
from lunarlink.utils import response
|
from lunarlink.utils.decorator import request_log
|
from lunarlink.utils.query_filters import filter_by_time_range, filter_by_node
|
from lunarlink.utils.parser import Format, Parse
|
from lunarlink.utils.enums.TreeTypeEnum import TreeType
|
|
|
class APITag(IntEnum):
|
DEPRECATED = 4 # 已废弃
|
|
|
class APITemplateView(GenericViewSet):
|
"""
|
API操作视图
|
"""
|
|
serializer_class = serializers.APISerializer
|
queryset = models.API.objects.filter(~Q(tag=APITag.DEPRECATED.value))
|
|
@swagger_auto_schema(query_serializer=serializers.AssertSerializer())
|
@method_decorator(request_log(level="DEBUG"))
|
def list(self, request):
|
"""
|
api-获取api列表
|
|
支持多种条件搜索
|
"""
|
|
ser = serializers.AssertSerializer(data=request.query_params)
|
if ser.is_valid():
|
node = ser.validated_data.get("node")
|
project = ser.validated_data.get("project")
|
search: str = ser.validated_data.get("search")
|
tag = ser.validated_data.get("tag")
|
rig_env = ser.validated_data.get("rigEnv")
|
showYAPI = ser.validated_data.get("showYAPI")
|
creator = ser.validated_data.get("creator")
|
start_time = ser.validated_data.get("start_time")
|
end_time = ser.validated_data.get("end_time")
|
|
queryset = (
|
self.get_queryset().filter(project=project).order_by("-create_time")
|
)
|
# 根据创建时间过滤
|
queryset = filter_by_time_range(queryset, start_time, end_time)
|
|
if creator:
|
queryset = queryset.filter(creator__name=creator)
|
|
if showYAPI is False:
|
queryset = queryset.filter(~Q(creator__name="yapi"))
|
|
if search != "":
|
search: List = search.split()
|
for key in search:
|
queryset = queryset.filter(
|
Q(name__contains=key) | Q(url__contains=key)
|
)
|
try:
|
queryset = filter_by_node(queryset, project, node, TreeType.API.value)
|
except RelationNotFound:
|
return Response(response.TREE_NOT_EXISTS)
|
|
if tag != "":
|
queryset = queryset.filter(tag=tag)
|
|
if rig_env != "":
|
queryset = queryset.filter(rig_env=rig_env)
|
|
pagination_queryset = self.paginate_queryset(queryset)
|
serializer = self.get_serializer(pagination_queryset, many=True)
|
paginated_response = self.get_paginated_response(serializer.data)
|
|
response_data = {
|
**response.API_GET_SUCCESS,
|
"data": paginated_response.data,
|
}
|
return Response(response_data)
|
else:
|
return Response(ser.errors, status=status.HTTP_400_BAD_REQUEST)
|
|
@method_decorator(request_log(level="INFO"))
|
def add(self, request):
|
"""
|
api-新增一个api
|
|
前端按照格式组装好，注意body
|
"""
|
|
api = Format(request.data)
|
api.parse()
|
|
api_body = {
|
"name": api.name,
|
"body": api.testcase,
|
"url": api.url,
|
"method": api.method,
|
"project": models.Project.objects.get(id=api.project),
|
"relation": api.relation,
|
"creator": request.user,
|
}
|
|
try:
|
models.API.objects.create(**api_body)
|
except DataError:
|
return Response(response.DATA_TO_LONG)
|
|
return Response(response.API_ADD_SUCCESS)
|
|
@swagger_auto_schema(
|
request_body=openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"header": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"header": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Header 对象"
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Header 描述"
|
),
|
},
|
),
|
"request": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"form": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"data": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Form data 对象"
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Form data 描述"
|
),
|
},
|
),
|
"json": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="JSON 对象"
|
),
|
"params": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"params": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Params 对象"
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Params 描述"
|
),
|
},
|
),
|
"files": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"files": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Files 对象"
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Files 描述"
|
),
|
},
|
),
|
},
|
),
|
"extract": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"extract": openapi.Schema(
|
type=openapi.TYPE_ARRAY,
|
items=openapi.Schema(type=openapi.TYPE_OBJECT),
|
description="Extract 数组",
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Extract 描述"
|
),
|
},
|
),
|
"validate": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"validate": openapi.Schema(
|
type=openapi.TYPE_ARRAY,
|
items=openapi.Schema(type=openapi.TYPE_OBJECT),
|
description="Validate 数组",
|
),
|
},
|
),
|
"variables": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"variables": openapi.Schema(
|
type=openapi.TYPE_ARRAY,
|
items=openapi.Schema(type=openapi.TYPE_OBJECT),
|
description="Variables 数组",
|
),
|
"desc": openapi.Schema(
|
type=openapi.TYPE_OBJECT, description="Variables 描述"
|
),
|
},
|
),
|
"hooks": openapi.Schema(
|
type=openapi.TYPE_OBJECT,
|
properties={
|
"setup_hooks": openapi.Schema(
|
type=openapi.TYPE_ARRAY,
|
items=openapi.Schema(type=openapi.TYPE_STRING),
|
description="Setup hooks 数组",
|
),
|
"teardown_hooks": openapi.Schema(
|
type=openapi.TYPE_ARRAY,
|
items=openapi.Schema(type=openapi.TYPE_STRING),
|
description="Teardown hooks 数组",
|
),
|
},
|
),
|
"url": openapi.Schema(type=openapi.TYPE_STRING, description="URL"),
|
"method": openapi.Schema(type=openapi.TYPE_STRING, description="请求方法"),
|
"name": openapi.Schema(type=openapi.TYPE_STRING, description="名称"),
|
"times": openapi.Schema(type=openapi.TYPE_INTEGER, description="次数"),
|
# 添加其他自定义字段
|
},
|
)
|
)
|
@method_decorator(request_log(level="INFO"))
|
def update(self, request, pk):
|
"""
|
api-更新单个api
|
|
更新单个api的内容
|
"""
|
api = Format(request.data)
|
api.parse()
|
|
api_body = {
|
"name": api.name,
|
"body": api.testcase,
|
"url": api.url,
|
"method": api.method,
|
"updater": request.user.id,
|
"update_time": timezone.now(),
|
}
|
|
objs = models.API.objects.filter(id=pk)
|
if objs:
|
objs.update(**api_body)
|
else:
|
return Response(response.API_NOT_FOUND)
|
|
return Response(response.API_UPDATE_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def move(self, request):
|
"""
|
api-批量更新api的目录
|
|
移动api到指定目录
|
"""
|
project: int = request.data.get("project")
|
relation: int = request.data.get("relation")
|
apis: List = request.data.get("api")
|
ids = [api["id"] for api in apis]
|
|
objs = models.API.objects.filter(project=project, id__in=ids)
|
if objs:
|
objs.update(
|
relation=relation,
|
updater=request.user.id,
|
update_time=timezone.now(),
|
)
|
else:
|
return Response(response.API_NOT_FOUND)
|
|
return Response(response.API_UPDATE_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def copy(self, request, pk):
|
"""
|
api-复制api
|
|
复制一个api
|
"""
|
name = request.data.get("name")
|
try:
|
api = models.API.objects.get(id=pk)
|
except ObjectDoesNotExist:
|
return Response(response.API_NOT_FOUND)
|
body = literal_eval(api.body)
|
body["name"] = name
|
api.body = body
|
api.id = None
|
api.name = name
|
api.creator = request.user
|
api.updater = request.user.id
|
api.save()
|
return Response(response.API_ADD_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def destroy(self, request, pk):
|
"""
|
api-删除单个api
|
|
pk: api id
|
"""
|
obj = models.API.objects.filter(id=pk)
|
if not obj:
|
return Response(response.API_NOT_FOUND)
|
|
obj.update(
|
is_deleted=True,
|
updater=request.user.id,
|
update_time=timezone.now(),
|
)
|
return Response(response.API_DEL_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def bulk_destroy(self, request):
|
"""
|
批量删除api
|
|
[{id:int}]
|
"""
|
ids = [content["id"] for content in request.data]
|
objs = models.API.objects.filter(Q(id__in=ids) & Q(is_deleted=False))
|
if not objs:
|
return Response(response.API_NOT_FOUND)
|
objs.update(
|
is_deleted=True,
|
update_time=timezone.now(),
|
updater=request.user.id,
|
)
|
|
return Response(response.API_DEL_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def add_tag(self, request):
|
"""
|
api-更新api的tag, 暂时默认为调式成功, 接口文档参数与实际传参不一致需优化
|
|
更新api的tag类型
|
"""
|
api_ids: List = request.data.get("api_ids", [])
|
|
if api_ids:
|
updated_rows = models.API.objects.filter(pk__in=api_ids).update(
|
tag=request.data.get("tag"),
|
update_time=timezone.now(),
|
updater=request.user.id,
|
)
|
if updated_rows > 0:
|
return Response(response.API_UPDATE_SUCCESS)
|
|
return Response(response.API_NOT_FOUND)
|
|
@method_decorator(request_log(level="INFO"))
|
def sync_case(self, request, pk):
|
"""
|
api-同步api到case_step
|
|
1.根据api_id查出("name", "body", "url", "method")
|
2.根据api_id更新case_step中的("name", "body", "url", "method", "updater")
|
3.更新case的update_time, updater
|
"""
|
source_api = (
|
models.API.objects.filter(pk=pk)
|
.values("name", "body", "url", "method", "project")
|
.first()
|
)
|
# 根据api反向查出project
|
project = source_api.pop("project")
|
|
project_case_ids = models.Case.objects.filter(project=project).values_list(
|
"id", flat=True
|
)
|
# 限制case_step只在当前项目
|
case_steps = models.CaseStep.objects.filter(
|
source_api_id=pk, case_id__in=project_case_ids
|
)
|
|
case_steps.update(
|
**source_api,
|
updater=request.user.id,
|
update_time=timezone.now(),
|
)
|
case_ids = case_steps.values_list("case", flat=True)
|
# 限制case只在当前项目
|
models.Case.objects.filter(pk__in=list(case_ids), project=project).update(
|
update_time=timezone.now(),
|
updater=request.user.id,
|
)
|
return Response(response.CASE_STEP_SYNC_SUCCESS)
|
|
@method_decorator(request_log(level="INFO"))
|
def single(self, request, pk):
|
"""
|
api-获取单个api详情，返回body信息
|
|
获取单个api的详细情况
|
"""
|
try:
|
api = models.API.objects.get(id=pk)
|
except ObjectDoesNotExist:
|
return Response(response.API_NOT_FOUND)
|
|
parse = Parse(literal_eval(api.body))
|
parse.parse_http()
|
|
resp = {
|
"id": api.id,
|
"body": parse.testcase,
|
"success": True,
|
"creator": api.creator.name,
|
"relation": api.relation,
|
"project": api.project.id,
|
}
|
|
return Response(resp)
|
