import httpx
import json
import logging
from typing import Dict, Any, Optional
from datetime import date, datetime
from config.settings import settings
logger = logging.getLogger(__name__)
class APIHelper:
"""API调用辅助类"""
def __init__(self):
self.base_url = settings.api_base_url
self.timeout = settings.api_timeout
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
if settings.api_key:
self.headers["Authorization"] = f"Bearer {settings.api_key}"
async def make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict[str, Any]] = None,
params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""发送HTTP请求"""
url = f"{self.base_url}{endpoint}"
try:
async with httpx.AsyncClient(timeout=self.timeout) as client:
response = await client.request(
method=method,
url=url,
json=data,
params=params,
headers=self.headers
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"HTTP错误: {e.response.status_code} - {e.response.text}")
raise Exception(f"API调用失败: {e.response.status_code}")
except httpx.RequestError as e:
logger.error(f"请求错误: {e}")
raise Exception(f"网络请求失败: {e}")
except Exception as e:
logger.error(f"未知错误: {e}")
raise Exception(f"请求处理失败: {e}")
def serialize_datetime(obj):
"""序列化datetime对象"""
if isinstance(obj, (datetime, date)):
return obj.isoformat()
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
def deserialize_datetime(data: Dict[str, Any]) -> Dict[str, Any]:
"""反序列化datetime字符串"""
for key, value in data.items():
if isinstance(value, str):
# 尝试解析日期时间
try:
if "T" in value and ":" in value:
data[key] = datetime.fromisoformat(value.replace("Z", "+00:00"))
elif "-" in value and len(value) == 10:
data[key] = datetime.strptime(value, "%Y-%m-%d").date()
except ValueError:
pass
return data
def format_response(success: bool, data: Any = None, message: str = None, error: str = None) -> Dict[str, Any]:
"""格式化API响应"""
response = {"success": success}
if data is not None:
response["data"] = data
if message:
response["message"] = message
if error:
response["error"] = error
return response
