# mipu-requests 米普代理 HTTP 客户端,兼容 `requests` 接口。 所有 HTTP 请求通过米普 API 转发,接口与 `requests` 完全一致,迁移只需改 `import` 和 `proxies`。 ## 安装 ```bash pip install mipu-requests ``` ## 快速开始 ```python import mipu_requests s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_client_key", client_secret="your_client_secret", agent="JQ", ) s.proxies = "res_mipu_bookxWdyg" resp = s.get("https://example.com/page") resp.raise_for_status() data = resp.json() ``` ## 从 requests 迁移 ```python # ── 迁移前 ────────────────────────── import requests s = requests.Session() s.proxies = {"https": "http://proxy:8080"} resp = s.get("https://example.com/api") # ── 迁移后 ────────────────────────── import mipu_requests s = mipu_requests.session( base_url="https://api-eu.mipuyun.com", client_key="your_key", client_secret="your_secret", agent="JQ", ) s.proxies = "res_mipu_bookxWdyg" resp = s.get("https://example.com/api") ``` 只需两步: 1. `import requests` → `import mipu_requests` 2. 构造时传入凭据,`proxies` 改为字符串 其余代码(`.get()`, `.post()`, `.json()`, `.raise_for_status()` 等)无需改动。 ## 请求方法 ```python resp = s.get(url, params={"key": "value"}, headers={...}) resp = s.post(url, data={"key": "value"}, headers={...}) resp = s.post(url, json={"key": "value"}, headers={...}) resp = s.put(url, data=body, headers={...}) resp = s.delete(url, headers={...}) ``` ## 异步模式 默认启用异步轮询,防止长连接断连: ```python # 默认异步 s = mipu_requests.session(..., async_mode=True) # 关闭异步 s = mipu_requests.session(..., async_mode=False) # 自定义轮询参数 s = mipu_requests.session( ..., poll_interval=3, # 轮询间隔(秒) poll_max=20, # 最大轮询次数 submit_timeout=60, # 提交超时(秒) poll_timeout=(5, 30), # 轮询超时 (connect, read) ) ``` ## Response 对象 与 `requests.Response` 完全兼容: ```python resp.status_code # int HTTP 状态码 resp.text # str 响应体文本 resp.content # bytes 响应体字节 resp.headers # dict 响应头 resp.ok # bool status_code 在 200-399 resp.url # str 请求 URL resp.json() # dict 解析 JSON resp.raise_for_status() # >=400 抛异常 ``` ## 异常处理 ```python from mipu_requests import RequestException, HTTPError, APIError, PollTimeoutError try: resp = s.get(url) resp.raise_for_status() except HTTPError as e: print(f"目标站错误: {e.response.status_code}") except APIError as e: print(f"API 错误: {e}") except PollTimeoutError as e: print(f"轮询超时: {e}") except RequestException as e: print(f"请求异常: {e}") ``` ## License MIT