RagFlow_Plugs/services/utils/http_client.py

import requests
from typing import Dict, Any, Optional

class HTTPClient:
    """HTTP请求工具类，用于发送各种HTTP请求"""
    
    def __init__(self, base_url: str, api_key: str = None):
        """
        初始化HTTP客户端
        
        Args:
            base_url: API基础URL
            api_key: API密钥（Bearer token）
        """
        self.base_url = base_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        
        # 设置默认请求头
        if self.api_key:
            self.session.headers.update({
                'Authorization': f'Bearer {self.api_key}'
            })
    
    def post(self, endpoint: str, data: Optional[Dict] = None, 
             json: Optional[Dict] = None, files: Optional[Dict] = None,
             headers: Optional[Dict] = None) -> Dict[str, Any]:
        """
        发送POST请求
        
        Args:
            endpoint: API端点路径（以/开头）
            data: 表单数据
            json: JSON数据
            files: 文件数据
            headers: 自定义请求头
            
        Returns:
            Dict: 响应JSON数据
        
        Raises:
            requests.exceptions.RequestException: 请求失败时抛出
        """
        url = f"{self.base_url}{endpoint}"
        response = self.session.post(
            url=url,
            data=data,
            json=json,
            files=files,
            headers=headers
        )
        response.raise_for_status()  # 抛出HTTP错误
        return response.json()
    
    def get(self, endpoint: str, params: Optional[Dict] = None,
            headers: Optional[Dict] = None) -> Dict[str, Any]:
        """
        发送GET请求
        
        Args:
            endpoint: API端点路径（以/开头）
            params: 查询参数
            headers: 自定义请求头
            
        Returns:
            Dict: 响应JSON数据
        
        Raises:
            requests.exceptions.RequestException: 请求失败时抛出
        """
        url = f"{self.base_url}{endpoint}"
        response = self.session.get(
            url=url,
            params=params,
            headers=headers
        )
        response.raise_for_status()  # 抛出HTTP错误
        return response.json()
    
    def upload_file(self, endpoint: str, file_path: str, file_field_name: str = 'file',
                   data: Optional[Dict] = None, headers: Optional[Dict] = None) -> Dict[str, Any]:
        """
        上传文件
        
        Args:
            endpoint: API端点路径（以/开头）
            file_path: 本地文件路径
            file_field_name: 表单字段名称
            data: 额外的表单数据
            headers: 自定义请求头
            
        Returns:
            Dict: 响应JSON数据
        
        Raises:
            requests.exceptions.RequestException: 请求失败时抛出
        """
        # 打开文件并构建files字典
        with open(file_path, 'rb') as f:
            files = {
                file_field_name: (file_path.split('/')[-1], f)
            }
            
            # 发送POST请求
            return self.post(endpoint, data=data, files=files, headers=headers)