everything-is-suitable/everything-is-suitable-test/python_e2e/core/api_client.py

"""
API客户端模块

提供HTTP请求封装和API调用功能。
"""

import requests
import json
from typing import Any, Dict, Optional, Callable
from dataclasses import dataclass
from enum import Enum


class HTTPMethod(Enum):
    """HTTP方法"""
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"


@dataclass
class APIResponse:
    """API响应"""
    status_code: int
    data: Any
    headers: Dict[str, str]
    success: bool
    error_message: Optional[str] = None


@dataclass
class APIRequest:
    """API请求"""
    method: HTTPMethod
    url: str
    headers: Optional[Dict[str, str]] = None
    params: Optional[Dict[str, Any]] = None
    data: Optional[Any] = None
    json_data: Optional[Dict[str, Any]] = None
    timeout: int = 30


class APIClient:
    """
    API客户端
    
    特性:
    - 支持多种HTTP方法
    - 自动JSON序列化/反序列化
    - 超时处理
    - 错误处理
    - 请求/响应拦截器
    """
    
    def __init__(
        self,
        base_url: str,
        default_headers: Optional[Dict[str, str]] = None,
        timeout: int = 30
    ):
        """
        初始化API客户端
        
        Args:
            base_url: 基础URL
            default_headers: 默认请求头
            timeout: 默认超时时间（秒）
        """
        self._base_url = base_url.rstrip('/')
        self._default_headers = default_headers or {}
        self._default_timeout = timeout
        self._request_interceptors: list = []
        self._response_interceptors: list = []
        self._session = requests.Session()
    
    def add_request_interceptor(self, interceptor: Callable[[APIRequest], APIRequest]) -> None:
        """添加请求拦截器"""
        self._request_interceptors.append(interceptor)
    
    def add_response_interceptor(self, interceptor: Callable[[APIResponse], APIResponse]) -> None:
        """添加响应拦截器"""
        self._response_interceptors.append(interceptor)
    
    def _build_url(self, endpoint: str) -> str:
        """构建完整URL"""
        endpoint = endpoint.lstrip('/')
        return f"{self._base_url}/{endpoint}"
    
    def _apply_request_interceptors(self, request: APIRequest) -> APIRequest:
        """应用请求拦截器"""
        for interceptor in self._request_interceptors:
            request = interceptor(request)
        return request
    
    def _apply_response_interceptors(self, response: APIResponse) -> APIResponse:
        """应用响应拦截器"""
        for interceptor in self._response_interceptors:
            response = interceptor(response)
        return response
    
    def request(self, api_request: APIRequest) -> APIResponse:
        """
        发送HTTP请求
        
        Args:
            api_request: API请求对象
            
        Returns:
            API响应对象
        """
        try:
            # 应用请求拦截器
            api_request = self._apply_request_interceptors(api_request)
            
            # 合并默认请求头
            headers = {**self._default_headers, **(api_request.headers or {})}
            
            # 发送请求
            response = self._session.request(
                method=api_request.method.value,
                url=api_request.url,
                headers=headers,
                params=api_request.params,
                data=api_request.data,
                json=api_request.json_data,
                timeout=api_request.timeout or self._default_timeout
            )
            
            # 解析响应
            try:
                response_data = response.json()
            except json.JSONDecodeError:
                response_data = response.text
            
            api_response = APIResponse(
                status_code=response.status_code,
                data=response_data,
                headers=dict(response.headers),
                success=200 <= response.status_code < 300
            )
            
            # 应用响应拦截器
            return self._apply_response_interceptors(api_response)
            
        except requests.exceptions.Timeout:
            return APIResponse(
                status_code=0,
                data=None,
                headers={},
                success=False,
                error_message="请求超时"
            )
        except requests.exceptions.ConnectionError as e:
            return APIResponse(
                status_code=0,
                data=None,
                headers={},
                success=False,
                error_message=f"连接错误: {str(e)}"
            )
        except Exception as e:
            return APIResponse(
                status_code=0,
                data=None,
                headers={},
                success=False,
                error_message=f"请求失败: {str(e)}"
            )
    
    def get(
        self,
        endpoint: str,
        params: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None
    ) -> APIResponse:
        """发送GET请求"""
        request = APIRequest(
            method=HTTPMethod.GET,
            url=self._build_url(endpoint),
            headers=headers,
            params=params,
            timeout=timeout or self._default_timeout
        )
        return self.request(request)
    
    def post(
        self,
        endpoint: str,
        json_data: Optional[Dict[str, Any]] = None,
        data: Optional[Any] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None
    ) -> APIResponse:
        """发送POST请求"""
        request = APIRequest(
            method=HTTPMethod.POST,
            url=self._build_url(endpoint),
            headers=headers,
            data=data,
            json_data=json_data,
            timeout=timeout or self._default_timeout
        )
        return self.request(request)
    
    def put(
        self,
        endpoint: str,
        json_data: Optional[Dict[str, Any]] = None,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None
    ) -> APIResponse:
        """发送PUT请求"""
        request = APIRequest(
            method=HTTPMethod.PUT,
            url=self._build_url(endpoint),
            headers=headers,
            json_data=json_data,
            timeout=timeout or self._default_timeout
        )
        return self.request(request)
    
    def delete(
        self,
        endpoint: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[int] = None
    ) -> APIResponse:
        """发送DELETE请求"""
        request = APIRequest(
            method=HTTPMethod.DELETE,
            url=self._build_url(endpoint),
            headers=headers,
            timeout=timeout or self._default_timeout
        )
        return self.request(request)