67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
import time
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from requests.packages.urllib3.util.retry import Retry
|
|
from typing import Optional, Dict, Any, Union
|
|
|
|
class HttpClient:
|
|
def __init__(self, max_retries: int = 3, backoff_factor: float = 0.5):
|
|
self.session = requests.Session()
|
|
# 配置重试策略
|
|
retry_strategy = Retry(
|
|
total=max_retries,
|
|
backoff_factor=backoff_factor,
|
|
status_forcelist=[500, 502, 503, 504, 429] # 这些状态码会触发重试
|
|
)
|
|
|
|
# 将重试策略应用到 session
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
self.session.mount("http://", adapter)
|
|
self.session.mount("https://", adapter)
|
|
|
|
def request(self,
|
|
method: str,
|
|
url: str,
|
|
headers: Optional[Dict] = None,
|
|
params: Optional[Dict] = None,
|
|
data: Optional[Union[Dict, str]] = None,
|
|
cookies: Optional[Dict] = None,
|
|
allow_redirects: bool = True,
|
|
timeout: int = 30,
|
|
**kwargs) -> requests.Response:
|
|
"""
|
|
通用请求方法
|
|
"""
|
|
try:
|
|
response = self.session.request(
|
|
method=method,
|
|
url=url,
|
|
headers=headers,
|
|
params=params,
|
|
data=data,
|
|
cookies=cookies,
|
|
allow_redirects=allow_redirects,
|
|
timeout=timeout,
|
|
**kwargs
|
|
)
|
|
response.raise_for_status()
|
|
return response
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"请求失败: {url}, 错误: {str(e)}")
|
|
raise
|
|
|
|
def get(self, url: str, **kwargs) -> requests.Response:
|
|
"""
|
|
GET 请求封装
|
|
"""
|
|
return self.request("GET", url, **kwargs)
|
|
|
|
def post(self, url: str, **kwargs) -> requests.Response:
|
|
"""
|
|
POST 请求封装
|
|
"""
|
|
return self.request("POST", url, **kwargs)
|
|
|
|
# 创建一个全局的 HttpClient 实例
|
|
http_client = HttpClient()
|