张翔
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
"""
|
|
基础API类
|
|
"""
|
|
|
|
from typing import Optional, Dict, Any
|
|
from httpx import AsyncClient, Response
|
|
from loguru import logger
|
|
|
|
|
|
class BaseAPI:
|
|
"""基础API类"""
|
|
|
|
def __init__(self, client: AsyncClient, base_url: str = ""):
|
|
self.client = client
|
|
self.base_url = base_url
|
|
|
|
async def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None, **kwargs) -> Response:
|
|
"""GET请求"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
logger.info(f"GET {url} - Params: {params}")
|
|
response = await self.client.get(url, params=params, **kwargs)
|
|
logger.info(f"Response: {response.status_code}")
|
|
return response
|
|
|
|
async def post(self, endpoint: str, data: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, **kwargs) -> Response:
|
|
"""POST请求"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
logger.info(f"POST {url} - Data: {data} - JSON: {json}")
|
|
response = await self.client.post(url, data=data, json=json, **kwargs)
|
|
logger.info(f"Response: {response.status_code}")
|
|
return response
|
|
|
|
async def put(self, endpoint: str, data: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, **kwargs) -> Response:
|
|
"""PUT请求"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
logger.info(f"PUT {url} - Data: {data} - JSON: {json}")
|
|
response = await self.client.put(url, data=data, json=json, **kwargs)
|
|
logger.info(f"Response: {response.status_code}")
|
|
return response
|
|
|
|
async def delete(self, endpoint: str, **kwargs) -> Response:
|
|
"""DELETE请求"""
|
|
url = f"{self.base_url}{endpoint}"
|
|
logger.info(f"DELETE {url}")
|
|
response = await self.client.delete(url, **kwargs)
|
|
logger.info(f"Response: {response.status_code}")
|
|
return response
|
|
|
|
async def assert_status_code(self, response: Response, expected_status: int):
|
|
"""断言状态码"""
|
|
assert response.status_code == expected_status, f"Expected {expected_status}, got {response.status_code}. Response: {response.text}"
|
|
|
|
async def assert_response_contains(self, response: Response, key: str, value: Any = None):
|
|
"""断言响应包含指定字段"""
|
|
data = response.json()
|
|
assert key in data, f"Response does not contain key '{key}'"
|
|
if value is not None:
|
|
assert data[key] == value, f"Expected {value}, got {data[key]}"
|