mipu-open/sdk

mipu-requests

米普代理 HTTP 客户端，兼容 requests 接口。

所有 HTTP 请求通过米普 API 转发，接口与 requests 完全一致，迁移只需改 import 和 proxies。

安装

pip install mipu-requests

快速开始

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 迁移

# ── 迁移前 ──────────────────────────
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() 等）无需改动。

请求方法

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={...})

异步模式

默认启用异步轮询，防止长连接断连：

# 默认异步
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 完全兼容：

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 抛异常

异常处理

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