python 的 httpx

### 使用 HTTPX 库进行流式读写操作 对于使用 `httpx` 进行流式请求，可以利用异步特性来处理大文件或其他需要高效管理资源的情况。下面展示了一个简单的例子，说明如何执行带有流响应的 GET 请求以及发送 POST 请求时的数据流。 #### 流式获取数据 (GET) 当发起 GET 请求并希望以流的方式接收服务器返回的内容时，可以通过设置参数 `stream=True` 来实现这一点： ```python import httpx with httpx.Client() as client: with client.stream('GET', 'https://example.com') as response: for chunk in response.iter_bytes(): print(chunk.decode()) # 处理接收到的数据块 ``` 此方法允许逐片下载远程资源而不将其全部加载到内存中[^1]。 #### 发送流式数据 (POST) 如果要上传大量数据或文件，则可以在构建请求体时提供可迭代对象作为 payload 的一部分，从而支持分批传输给定的信息至目标 URL: ```python def generate_data(): yield b'--boundary\r\n' yield b'Content-Disposition: form-data; name="field"\r\n\r\n' yield b'value\r\n' yield b'--boundary--\r\n' async def main(): async with httpx.AsyncClient() as client: await client.post( url='https://example.com/upload', content=generate_data(), headers={'content-type': 'multipart/form-data; boundary=----boundary'} ) if __name__ == '__main__': import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(main()) ``` 上述代码片段展示了怎样通过自定义函数生成待上传的数据序列，并将其传递给 `post()` 方法中的 `content` 参数完成流式的 POST 请求[^2]。 请注意，在实际应用过程中应当根据具体需求调整协议版本、超时配置以及其他必要的选项。

### 使用 Python 的 `httpx` 库进行 HTTP 请求 `httpx` 是一个功能强大且现代的 Python HTTP 客户端库，支持同步和异步请求，并兼容 HTTP/1.1 和 HTTP/2 协议。该库设计简洁、易于使用，适用于构建 RESTful API 客户端、爬虫系统以及微服务通信等场景。 #### 1. 发起基本的 GET 请求 `httpx` 提供了类似 `requests` 的接口，允许开发者快速发起 HTTP 请求并处理响应内容。以下是一个基本的 GET 请求示例： ```python import httpx response = httpx.get("https://httpbin.org/get") print(response.status_code) # 输出状态码 print(response.json()) # 输出 JSON 格式的响应数据 ``` GET 请求可以附加查询参数，通过 `params` 参数传递字典形式的数据： ```python params = {"key1": "value1", "key2": "value2"} response = httpx.get("https://httpbin.org/get", params=params) ``` #### 2. 发起 POST 请求 POST 请求通常用于向服务器提交数据，`httpx` 支持多种数据格式，包括表单数据、JSON 数据等。例如，发送 JSON 数据可以使用 `json` 参数： ```python data = {"username": "test", "password": "123456"} response = httpx.post("https://httpbin.org/post", json=data) print(response.json()) ``` 对于表单数据，可以使用 `data` 参数代替 `json` 参数： ```python form_data = {"name": "Alice", "age": "30"} response = httpx.post("https://httpbin.org/post", data=form_data) ``` #### 3. 设置请求头 在某些情况下，需要自定义请求头以满足服务器的要求。`httpx` 允许通过 `headers` 参数指定自定义头部信息： ```python headers = {"User-Agent": "MyApp/1.0", "Authorization": "Bearer token123"} response = httpx.get("https://httpbin.org/get", headers=headers) ``` #### 4. 处理超时与重试机制 默认情况下，`httpx` 的请求会设置一个合理的超时时间（通常是 5 秒）。可以通过 `timeout` 参数调整超时时间： ```python response = httpx.get("https://httpbin.org/get", timeout=10.0) ``` 为了增强健壮性，可以在请求中加入重试逻辑。虽然 `httpx` 本身不直接提供重试功能，但可以结合 `tenacity` 等第三方库实现： ```python from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(3)) def fetch(): return httpx.get("https://httpbin.org/get", timeout=5.0) response = fetch() ``` #### 5. 异步请求支持 `httpx` 支持异步请求，适合高并发的 I/O 密集型任务。异步请求需使用 `AsyncClient` 类： ```python import asyncio import httpx async def fetch(): async with httpx.AsyncClient() as client: response = await client.get("https://httpbin.org/get") print(response.json()) asyncio.run(fetch()) ``` #### 6. 文件上传与下载 `httpx` 支持文件上传，通过 `files` 参数指定要上传的文件： ```python with open("example.txt", "rb") as f: files = {"file": ("example.txt", f, "text/plain")} response = httpx.post("https://httpbin.org/post", files=files) ``` 对于文件下载，可以直接将响应内容写入本地文件： ```python with open("downloaded_file.txt", "wb") as f: with httpx.stream("GET", "https://example.com/file.txt") as response: for chunk in response.iter_bytes(): f.write(chunk) ``` #### 7. 认证与 Cookie 管理 `httpx` 支持基本认证（Basic Auth）和 Bearer Token 认证。基本认证可以通过 `auth` 参数指定用户名和密码： ```python response = httpx.get("https://httpbin.org/basic-auth/user/passwd", auth=("user", "passwd")) ``` Bearer Token 认证则可以通过在请求头中添加 `Authorization` 字段实现： ```python headers = {"Authorization": "Bearer your_token_here"} response = httpx.get("https://api.example.com/data", headers=headers) ``` 此外，`httpx` 还提供了 `Client` 类用于管理会话中的 Cookie： ```python with httpx.Client() as client: client.get("https://httpbin.org/cookies/set?name=value") response = client.get("https://httpbin.org/cookies") print(response.json()) ```