🎈 Perf(optimized code): 优化代码

Author: RanY-Luck

backend/plugin/api_testing/api/v1/assertion.py

@@ -4,21 +4,18 @@
 断言API
 """
 from typing import Any, Dict, List
-
 from fastapi import APIRouter, Body
-from fastapi.responses import JSONResponse
-
-from backend.common.response.response_schema import response_base
-from backend.plugin.api_testing.utils.assertion import Assertion, AssertionEngine, AssertionResult
+from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
+from backend.plugin.api_testing.utils.assertion import Assertion, AssertionEngine
 
 router = APIRouter()
 
 
-@router.post("/validate", response_model=Dict[str, Any], summary="执行断言验证")
+@router.post("/validate", response_model=ResponseModel, summary="执行断言验证")
 async def execute_assertion(
-    assertion: Assertion = Body(...),
-    response_data: Dict[str, Any] = Body(...)
-) -> JSONResponse:
+        assertion: Assertion = Body(...),
+        response_data: Dict[str, Any] = Body(...)
+) -> ResponseModel | ResponseSchemaModel:
     """
     执行断言验证接口
     
@@ -27,17 +24,17 @@ async def execute_assertion(
     try:
         # 执行断言
         result = AssertionEngine.execute_assertion(assertion, response_data)
-        
+
         return response_base.success(data=result.model_dump())
     except Exception as e:
-        return response_base.fail(msg=f"断言执行失败: {str(e)}")
+        return response_base.fail(data=f"断言执行失败: {str(e)}")
 
 
-@router.post("/batch-validate", response_model=Dict[str, Any], summary="批量执行断言验证")
+@router.post("/batch-validate", response_model=ResponseModel, summary="批量执行断言验证")
 async def execute_batch_assertions(
-    assertions: List[Assertion] = Body(...),
-    response_data: Dict[str, Any] = Body(...)
-) -> JSONResponse:
+        assertions: List[Assertion] = Body(...),
+        response_data: Dict[str, Any] = Body(...)
+) -> ResponseModel | ResponseSchemaModel:
     """
     批量执行断言验证接口
     
@@ -46,7 +43,7 @@ async def execute_batch_assertions(
     try:
         # 执行批量断言
         results = AssertionEngine.execute_assertions(assertions, response_data)
-        
+
         # 构建响应
         data = {
             "results": [result.model_dump() for result in results],
@@ -56,7 +53,7 @@ async def execute_batch_assertions(
                 "failed": sum(1 for result in results if not result.success)
             }
         }
-        
+
         return response_base.success(data=data)
     except Exception as e:
-        return response_base.fail(msg=f"批量断言执行失败: {str(e)}")
+        return response_base.fail(data=f"批量断言执行失败: {str(e)}")

backend/plugin/api_testing/api/v1/data_driven.py

@@ -3,19 +3,13 @@
 """
 数据驱动测试API
 """
-from typing import Dict, List, Optional, Any, Coroutine
-
-from fastapi import APIRouter, Body, Path, Query, UploadFile, File, Form
-from fastapi.responses import JSONResponse
 import pandas as pd
 import json
 import os
-import tempfile
-
+from fastapi import APIRouter, Path, UploadFile, File, Form
 from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
 from backend.plugin.api_testing.utils.data_driven import (
-    DataDriverManager, DataDrivenConfig, DataSourceConfig,
-    DataSourceType, DatabaseType, TestCaseParameter, TestIteration
+    DataDriverManager, DataDrivenConfig, DataSourceConfig, DataSourceType
 )
 
 router = APIRouter()

backend/plugin/api_testing/api/v1/environment.py

@@ -3,15 +3,12 @@
 """
 环境和变量管理API
 """
-from typing import Dict, List, Optional, Any
-
+from typing import Dict, Optional, Any
 from fastapi import APIRouter, Body, Path, Query
-from fastapi.responses import JSONResponse
 
 from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
 from backend.plugin.api_testing.utils.environment import (
-    EnvironmentManager, EnvironmentModel, VariableManager, VariableModel,
-    VariableScope, EnvironmentType
+    EnvironmentManager, EnvironmentModel, VariableManager, VariableModel, VariableScope
 )
 
 router = APIRouter()

backend/plugin/api_testing/api/v1/history.py

@@ -5,9 +5,8 @@
 提供接口请求历史记录的API
 """
 from datetime import datetime
-from typing import Any, Dict, List, Optional, Coroutine
+from typing import Any, List, Optional
 from fastapi import APIRouter, Body, Query
-from fastapi.responses import JSONResponse
 from pydantic import BaseModel, Field
 from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
 from backend.plugin.api_testing.utils.history import HistoryManager, RequestHistoryItem

backend/plugin/api_testing/api/v1/report.py

@@ -3,22 +3,20 @@
 """
 测试报告API
 """
-from typing import Any, Dict, List
-
+from typing import Any, Dict
 from fastapi import APIRouter, Body
-from fastapi.responses import JSONResponse, HTMLResponse
-
-from backend.common.response.response_schema import response_base
+from fastapi.responses import HTMLResponse
+from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
 from backend.plugin.api_testing.utils.report_generator import ReportFormat, TestReport, report_generator
 
 router = APIRouter()
 
 
 @router.post("/generate", response_model=Dict[str, Any], summary="生成测试报告")
 async def generate_report(
-    report_data: TestReport = Body(...),
-    format: ReportFormat = Body(ReportFormat.HTML)
-) -> JSONResponse:
+        report_data: TestReport = Body(...),
+        format: ReportFormat = Body(ReportFormat.HTML)
+) -> ResponseModel | ResponseSchemaModel:
     """
     生成测试报告接口
     
@@ -27,18 +25,18 @@ async def generate_report(
     try:
         # 生成报告
         report_content = report_generator.generate_report(report_data, format)
-        
+
         # 构建响应
         data = {"content": report_content, "format": format}
-        
+
         return response_base.success(data=data)
     except Exception as e:
-        return response_base.fail(msg=f"报告生成失败: {str(e)}")
+        return response_base.fail(data=f"报告生成失败: {str(e)}")
 
 
 @router.post("/preview", response_class=HTMLResponse, summary="预览HTML测试报告")
 async def preview_html_report(
-    report_data: TestReport = Body(...)
+        report_data: TestReport = Body(...)
 ) -> HTMLResponse:
     """
     预览HTML测试报告
@@ -48,7 +46,7 @@ async def preview_html_report(
     try:
         # 生成HTML报告
         report_content = report_generator.generate_report(report_data, ReportFormat.HTML)
-        
+
         # 直接返回HTML内容
         return HTMLResponse(content=report_content)
     except Exception as e:
@@ -60,4 +58,4 @@ async def preview_html_report(
             </body>
         </html>
         """
-        return HTMLResponse(content=error_html, status_code=500)
+        return HTMLResponse(content=error_html, status_code=500)

backend/plugin/api_testing/model/models.py

@@ -1,13 +1,11 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 from datetime import datetime
-from typing import Dict, List, Optional, Any
-
+from typing import List
 from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean, JSON, ForeignKey
-from sqlalchemy.orm import relationship
-
+from sqlalchemy.orm import relationship, Mapped
 from backend.common.enums import StatusType
-from backend.database.db_mysql import Base
+from backend.common.model import Base
 
 
 class ApiProject(Base):
@@ -25,7 +23,7 @@ class ApiProject(Base):
     update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='更新时间')
 
     # 关联关系
-    test_cases = relationship("ApiTestCase", back_populates="project")
+    test_cases: Mapped[List["ApiTestCase"]] = relationship("ApiTestCase", back_populates="project")
 
 
 class ApiTestCase(Base):
@@ -43,9 +41,9 @@ class ApiTestCase(Base):
     update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='更新时间')
 
     # 关联关系
-    project = relationship("ApiProject", back_populates="test_cases")
-    steps = relationship("ApiTestStep", back_populates="test_case")
-    reports = relationship("ApiTestReport", back_populates="test_case")
+    project: Mapped[List["ApiProject"]] = relationship("ApiProject", back_populates="test_cases")
+    steps: Mapped[List["ApiTestStep"]] = relationship("ApiTestStep", back_populates="test_case")
+    reports: Mapped[List["ApiTestReport"]] = relationship("ApiTestReport", back_populates="test_case")
 
 
 class ApiTestStep(Base):
@@ -74,7 +72,7 @@ class ApiTestStep(Base):
     update_time = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='更新时间')
 
     # 关联关系
-    test_case = relationship("ApiTestCase", back_populates="steps")
+    test_case: Mapped[List["ApiTestCase"]] = relationship("ApiTestCase", back_populates="steps")
 
 
 class ApiTestReport(Base):
@@ -95,4 +93,4 @@ class ApiTestReport(Base):
     create_time = Column(DateTime, default=datetime.now, comment='创建时间')
 
     # 关联关系
-    test_case = relationship("ApiTestCase", back_populates="reports")
+    test_case: Mapped[List["ApiTestCase"]] = relationship("ApiTestCase", back_populates="reports")