Skip to content

SDK Usage

Jump to bottom
Hades edited this page Aug 23, 2026 · 5 revisions

SDK 用法

Python 客户端的完整用法。只想快速跑通看快速开始， 想照着现成场景改看示例集，要查签名和字段看API 参考。

前提：服务端已经在跑（ipclick run）。客户端只是把任务通过 gRPC 交给它。

建一个客户端

三种方式，按"要不要自己管生命周期"来选。

1. with Downloader() —— 推荐

from ipclick import Downloader

with Downloader() as d:
    print(d.get("https://example.com").status_code)
# 退出 with 时自动关闭 gRPC 连接

不传参数就按配置文件连（默认 127.0.0.1:9528）。要显式指定：

with Downloader(host="10.0.0.5", port=9528, token="...") as d:
    ...

with Downloader(config_path="/etc/ipclick/ipclick.toml") as d:
    ...

2. 进程级共享单例

不想在每个函数里传客户端时用它。同一组连接参数只建一个，进程内复用：

from ipclick import get_downloader, close_all_downloaders

d = get_downloader()                  # 第一次建，后面拿同一个
d2 = get_downloader()                 # d2 is d
...
close_all_downloaders()               # 进程退出前统一关

3. 模块级懒加载对象

最省事的写法，导入即用，第一次调用时才真的建连接：

from ipclick import downloader

resp = downloader.get("https://example.com")

适合脚本和 REPL。长期运行的服务里建议用前两种——生命周期看得见。

create_client() 会按 [GENERAL].mode 决定给你单机客户端还是集群客户端 （ClusterDownloader）。想让"单机还是集群"由配置说话、代码不改，就用它。

基本请求

with Downloader() as d:
    d.get("https://example.com")
    d.post("https://api.example.com/items", json={"name": "x"})
    d.put("https://api.example.com/items/1", data=b"raw bytes")
    d.patch("https://api.example.com/items/1", json={"name": "y"})
    d.delete("https://api.example.com/items/1")
    d.head("https://example.com")
    d.options("https://example.com")

get() 的第二个位置参数是 query string：

d.get("https://httpbin.org/get", {"page": 2, "q": "关键词"})
# 等价于 params={"page": 2, "q": "关键词"}

post() 的 data 与 json 是两回事：

d.post(url, json={"a": 1})        # Content-Type: application/json，自动序列化
d.post(url, data=b"a=1&b=2")      # 原样发送 bytes，Content-Type 自己在 headers 里给
d.post(url, data="纯文本")         # str 会按 UTF-8 编码

请求头与 Cookie

resp = d.get(
    "https://example.com",
    headers={"User-Agent": "MyBot/1.0", "Accept-Language": "zh-CN"},
    cookies={"session": "abc123"},
)

cookies 也接受原始 Cookie 头字符串：

d.get(url, cookies="session=abc123; theme=dark")

不传 User-Agent 时会自动填一个真实浏览器的。要保持默认伪装就不要覆盖它—— 自己填一个 python-requests/2.x 反而更容易被拦。

request() —— 一次给全所有参数

便捷方法是它的薄包装。参数多的时候直接用它更清楚：

from ipclick import Downloader, HttpMethod

with Downloader() as d:
    resp = d.request(
        method=HttpMethod.POST,
        url="https://api.example.com/search",
        headers={"Authorization": "Bearer ..."},
        params={"page": 1},
        json={"q": "关键词"},
        timeout=30,
        max_retries=3,
        retry_backoff=2.0,
        allow_redirects=True,
        verify=True,
        adapter="curl_cffi",
    )

method 必须传 HttpMethod 枚举成员（如 HttpMethod.POST）——不像 adapter 那样接受字符串；传 "POST" 这样的字符串会在发送时抛出 ValidationError。 完整参数表见API 参考。

读响应

resp = d.get("https://httpbin.org/json")

resp.status_code      # int，-1 表示没拿到 HTTP 响应
resp.ok               # bool，等价于 is_success()
resp.text             # str
resp.content          # bytes
resp.json()           # dict/list，解析失败抛异常
resp.headers          # dict[str, str]
resp.elapsed_ms       # int，服务端测的耗时
resp.url              # 最终 URL（跟完重定向）
resp.error            # str | None
resp.request_uuid     # 这次请求的 id，拿它去查链路

raise_for_status() 在失败时抛异常，适合"出错就该崩"的脚本：

resp = d.get(url)
resp.raise_for_status()      # 4xx/5xx 或传输失败都会抛
data = resp.json()

resp.trace —— 这次请求实际是怎么跑的

t = resp.trace
t.node_id       # 哪台机器执行的
t.adapter       # 实际用了哪个适配器
t.attempts      # 试了几次（1 = 一次就成）
t.forwarded     # 是不是被转发到别的节点
t.queued_ms     # 在限流闸门里排了多久

排查"为什么慢"先看 queued_ms 和 attempts：前者大是被自己的限流挡住了， 后者大是目标站点在抖。

错误处理

核心规则：网络问题不抛异常，用法问题抛异常。

resp = d.get("https://does-not-exist.invalid")
# 不抛异常
print(resp.status_code)   # -1
print(resp.error)         # 具体错误
print(resp.ok)            # False

所以正常的判断长这样：

resp = d.get(url)
if not resp.ok:
    log.warning("抓取失败 %s: %s", url, resp.error)
    return None
return resp.text

而这些情况会抛异常：

from ipclick import ValidationError, AuthenticationError, AdapterError, URLNotAllowedError

d.get("")                              # ValidationError：URL 空
d.get("file:///etc/passwd")            # ValidationError：前缀不是 http:// / https://
d.get("http://169.254.169.254/")       # URLNotAllowedError：服务端 SSRF 准入拒绝
d.get(url, adapter="camoufox")         # AdapterError：这个适配器没装
# 令牌不对                              # AuthenticationError

注意上面第二、三行的区别。 客户端只看 URL 前缀，file:// 连不上服务端就被 ValidationError 挡了；协议白名单、内网地址、云元数据地址这些策略全在服务端， 要等服务端回话才会抛 URLNotAllowedError。所以只写 except URLNotAllowedError 接不住 file:// 那一条。

异常层次见 API 参考。想统一兜住就抓基类 IPClickError。

指定适配器

d.get(url)                          # 默认 curl_cffi，带 TLS 指纹伪装
d.get(url, adapter="niquests")      # HTTP/2 + HTTP/3
d.get(url, adapter="camoufox")      # 真浏览器渲染
d.get(url, adapter="browser")       # 按平台自动挑一个浏览器引擎

六个适配器怎么选见适配器。指定没装的适配器会抛 AdapterError 并带上安装命令，不会静默换一个。

指纹伪装

curl_cffi 可以点名模仿某个浏览器版本：

d.get(url, impersonate="chrome131")

不传时客户端会自动填 impersonate="chrome"，curl_cffi 再把它解析成当前版本的 Chrome 指纹。没有对应的配置项——改 ipclick.toml 改不动它，要点名版本只能按请求传 impersonate=。可选值随 curl_cffi 版本变化，见适配器。

代理

三种写法：

d.get(url, proxy=True)                            # 用配置文件里的 [PROXY]
d.get(url, proxy="http://user:pass@1.2.3.4:8080") # 直接给 URL
from ipclick import ProxyConfig

d.get(url, proxy=ProxyConfig(
    scheme="http", host="1.2.3.4", port=8080,
    auth_key="user", auth_password="pass",
))

proxy=True 但配置里没填代理时会打一条警告并直连——不会静默当成"用了代理"。 账号密码建议走 .env（IPCLICK_PROXY_AUTH_KEY / IPCLICK_PROXY_AUTH_PASSWORD）， 见配置体系。

浏览器渲染

拿渲染后的 DOM，而不是原始 HTML：

resp = d.get("https://spa.example.com", adapter="camoufox")
print(resp.text)      # JS 跑完之后的 DOM

等待策略、拦截资源、超时这些用 automation_config（JSON 字符串）：

import json

resp = d.get(
    "https://spa.example.com",
    adapter="camoufox",
    automation_config=json.dumps({
        "wait_until": "networkidle",
        "block_resources": ["image", "font", "media"],
        "wait_for_selector": "#content",
        "wait_for_timeout": 3000,
    }),
)

block_resources 是最有效的提速手段——不下图片和字体通常能省掉一半时间。 完整参数与调优见浏览器渲染。

在页面里执行 JS

script = """async () => {
    document.querySelector('#login').click();
    await new Promise(r => setTimeout(r, 1000));
    return document.title;
}"""

resp = d.get(url, adapter="camoufox", automation_script=script)

脚本有两种写法，但只有一种能用 await：直接写函数体（return document.title;） 时，它会被包成一个同步箭头函数 () => { ... }，函数体里出现 await 就是 JS 语法错（会作为 ValidationError 报回来）。要用 await 就得像上面那样自己写成 async () => {...} 或 async function () {...} 开头——以 function / async / ( / => 开头的脚本会原样透传，不再包装。

⚠️ automation_script 绕过 SSRF 准入——脚本里可以自己发请求到任何地址。 它默认关闭（[BROWSER].allow_scripts），只在你信任调用方时打开。 见安全。

流式下载

响应体不进内存，边收边处理。适合大文件：

with Downloader() as d:
    with d.stream("https://example.com/big.zip") as s:
        print(s.status_code, s.content_length)
        with open("big.zip", "wb") as f:
            for chunk in s:
                f.write(chunk)
        print(s.total_bytes, s.elapsed_ms)

要点：

  • 一定要用 with（或手动 close()）。提前 break 出循环时，with 会取消 底层 RPC；不关的话服务端那次抓取还在跑。
  • s.total_bytes / s.elapsed_ms 要消费完才有值——它们在流末尾的 trailer 里。
  • s.is_success() 在消费完之后才计入 trailer 里的错误。

断点续传

网络断了自动用 Range 续传，不重下已经拿到的部分：

from ipclick import Downloader
from ipclick.resume import download_to_file

with Downloader() as d:
    result = download_to_file(d, "https://example.com/big.iso", "big.iso")
    print(result.total_bytes, result.attempts, result.restarts)

attempts > 1 说明中途断过；restarts > 0 说明没能续上、从头重下过—— 服务端没给 206，或者压根不支持 Range，之前下的字节全白费了。 两个都为真时是"断过且重下过"，只有 attempts > 1 and restarts == 0 才是 "断过并真的续上了"。带进度回调：

def on_chunk(received: int, total: int) -> None:
    pct = received / total * 100 if total else 0
    print(f"\r{pct:.1f}%", end="")

download_to_file(d, url, "big.iso", chunk_callback=on_chunk, max_attempts=10)

只要字节流不要落盘用 iter_resumable：

from ipclick.resume import iter_resumable

for chunk in iter_resumable(d, url):
    process(chunk)

服务端不支持 Range 时会从头重下（并在日志里说明），不会静默给你一个残缺文件。

批量请求

一次 RPC 发多个任务，按完成顺序返回——快的先回，不用等最慢那个：

import uuid_utils as uuid
from ipclick import Downloader, DownloadTask

tasks = [DownloadTask(uuid=str(uuid.uuid7()), url=f"https://example.com/p/{i}") for i in range(100)]

with Downloader() as d:
    for resp in d.batch(tasks):
        print(resp.request_uuid, resp.url, resp.status_code)

每个任务要自己填 uuid。 DownloadTask.uuid 默认是空字符串——真正发出去的 uuid 只在 to_protobuf()（也就是发送那一刻）才生成，不会写回你手上的这个 DownloadTask 对象。按完成顺序收响应、想用 resp.request_uuid 认出它对应哪个 任务时，就必须像上面这样在构造 DownloadTask 时自己传 uuid=；不传的话事后 没有任何办法拿到服务端真正用的那个 uuid。

提前 break 出循环是安全的：那条双向流会被取消，服务端不再继续替你把余下的任务打出去。 这一步取消是必需的——batch() 默认 timeout=None，也就是这条流没有 deadline， 半开着挂在那里的话，服务端会一直守着流和未取走的任务，直到连接断开。

DownloadTask 接受和 request() 一样的参数，可以每个任务不同：

tasks = [
    DownloadTask(uuid=str(uuid.uuid7()), url="https://a.example.com", adapter="curl_cffi", timeout=10),
    DownloadTask(uuid=str(uuid.uuid7()), url="https://b.example.com", adapter="camoufox", timeout=60),
]

注意：批量整批发给同一个节点，不跨节点拆分。要打散到多个节点就自己分批， 或者用集群的客户端分发模式。

异步客户端

AsyncDownloader 不在顶层导出，从 ipclick.aio 导入：

import asyncio
from ipclick.aio import AsyncDownloader

async def main():
    async with AsyncDownloader() as d:
        resp = await d.get("https://example.com")
        print(resp.status_code)

asyncio.run(main())

并发抓一批：

async def fetch_all(urls: list[str]) -> list[str]:
    async with AsyncDownloader() as d:
        results = await asyncio.gather(*(d.get(u) for u in urls))
        return [r.text for r in results if r.ok]

流式与批量都有异步版：

async with AsyncDownloader() as d:
    async with await d.stream(url) as s:
        async for chunk in s:
            ...

    async for resp in d.batch(tasks):
        ...

同步和异步两条流式路径的语义是一致的——都会在提前退出时取消底层 RPC。

channel 绑事件循环：AsyncDownloader 的连接建立在第一次请求所在的事件循环上。 不要跨 asyncio.run() 复用同一个实例。

关闭与复用

gRPC 连接建起来就该复用——每个请求新建一个客户端会把连接握手的开销加到每次请求上。

# 好：一个客户端打很多请求
with Downloader() as d:
    for url in urls:
        d.get(url)

# 差：每次都新建
for url in urls:
    with Downloader() as d:
        d.get(url)

长期运行的服务里用 get_downloader() 拿进程级单例，退出时 close_all_downloaders()。

客户端关掉之后再用会抛 ClientClosedError——不会静默重连，因为那会掩盖 "你在用一个本该已经释放的对象"这个问题。

下一步

IPClick

开始

用法

能力

运维

参考

Clone this wiki locally