当前位置：物联沃-IOTWORD物联网 > 技术教程 > 【Python】httpx库详解：HTTP客户端请求与响应处理指南
代码收藏家代码收藏家 技术教程

【Python】httpx库详解：HTTP客户端请求与响应处理指南

httpx 是一个现代化的 Python HTTP 客户端库，设计用于发送 HTTP 请求和处理响应。它支持同步和异步 API，兼容 requests 库的接口，同时提供更强大的功能，如 HTTP/2、连接池、流式响应和异步支持。httpx 适合构建高性能 Web 客户端、爬虫、API 交互工具，以及需要异步 I/O 的应用。

以下是对 httpx 库的详细介绍，包括其功能、用法和实际应用。

1. httpx 库的作用

  • HTTP 请求：支持 GET、POST、PUT、DELETE 等 HTTP 方法。
  • 异步支持：提供异步 API，适合高并发场景（如爬虫、微服务）。
  • HTTP/2：支持 HTTP/2 协议，提升性能。
  • 连接池：自动管理连接复用，减少开销。
  • 流式响应：支持处理大文件或实时数据流。
  • 兼容性：与 requests API 类似，易于迁移。
  • 扩展性：支持自定义头、认证、超时、代理等。

    • 2. 安装与环境要求

  • Python 版本：支持 Python 3.8+（推荐 3.9+）。
  • 依赖：
  • httpcore：核心 HTTP 处理。
  • certifi：SSL 证书验证。
  • 可选：h2（HTTP/2 支持）、brotli（Brotli 压缩）。
  • 安装命令： 
    pip install httpx
  • 完整功能（包括 HTTP/2 和 CLI）： 
    pip install httpx[http2,cli]
  • 异步支持：
  • 默认支持异步，无需额外安装（依赖 anyio 或 asyncio）。
  • 验证安装： 
    import httpx
print(httpx.__version__)  # 示例输出: 0.27.2

    • 3. 核心功能与用法

    httpx 提供同步 (httpx.Client) 和异步 (httpx.AsyncClient) 接口，核心类为 Client 和 AsyncClient。以下是主要功能和示例。

    3.1 基本请求（同步）

    发送简单的 HTTP 请求，类似 requests。

    import httpx

# 发送 GET 请求
response = httpx.get("https://api.example.com/data")
print(response.status_code)  # 输出: 200
print(response.json())  # 输出 JSON 数据

# 发送 POST 请求
data = {"key": "value"}
response = httpx.post("https://api.example.com/submit", json=data)
print(response.text)

    说明：

  • httpx.get、httpx.post 等是便捷方法，自动创建临时客户端。
  • response.json() 解析 JSON 响应。
  • 支持 params（查询参数）、headers、json 等参数。
    • 3.2 异步请求

    使用 AsyncClient 进行异步 HTTP 请求。

    import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        print(response.status_code)
        print(response.json())

# 运行异步代码
asyncio.run(main())

    说明：

  • AsyncClient 使用 async/await 语法，适合高并发。
  • async with 确保客户端正确关闭。
    • 3.3 使用 Client（连接池）

    通过 Client 或 AsyncClient 复用连接，提高性能。

    import httpx

# 同步客户端
with httpx.Client() as client:
    response1 = client.get("https://api.example.com/data1")
    response2 = client.get("https://api.example.com/data2")
    print(response1.json(), response2.json())

# 异步客户端
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        response1 = await client.get("https://api.example.com/data1")
        response2 = await client.get("https://api.example.com/data2")
        print(response1.json(), response2.json())

asyncio.run(main())

    说明：

  • Client 维护连接池，减少重复建立连接的开销。
  • with 语句确保客户端正确关闭。
    • 3.4 流式响应

    处理大文件或实时数据流。

    import httpx

with httpx.stream("GET", "https://example.com/large-file") as response:
    for chunk in response.iter_bytes():
        print(chunk[:100])  # 处理数据块

    异步流式响应：

    import httpx
import asyncio

async def main():
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", "https://example.com/large-file") as response:
            async for chunk in response.aiter_bytes():
                print(chunk[:100])

asyncio.run(main())

    说明：

  • stream 启用流式响应，适合下载大文件或处理实时数据。
  • iter_bytes/aiter_bytes 按块读取响应。
    • 3.5 自定义配置

    支持超时、头信息、代理、认证等。

    import httpx

# 自定义超时和头信息
headers = {"User-Agent": "MyApp/1.0"}
timeout = httpx.Timeout(10.0, connect=5.0)
response = httpx.get("https://api.example.com", headers=headers, timeout=timeout)
print(response.status_code)

# 代理
proxies = {"https://": "http://proxy.example.com:8080"}
response = httpx.get("https://api.example.com", proxies=proxies)

# 认证
auth = httpx.BasicAuth(username="user", password="pass")
response = httpx.get("https://api.example.com", auth=auth)

    说明：

  • Timeout：设置连接、读取等超时。
  • proxies：支持 HTTP/HTTPS 代理。
  • auth：支持基本认证、Digest 认证等。
    • 3.6 错误处理

    捕获 HTTP 和网络错误。

    import httpx

try:
    response = httpx.get("https://nonexistent.example.com", timeout=5.0)
    response.raise_for_status()  # 抛出 HTTP 错误
except httpx.HTTPStatusError as e:
    print(f"HTTP error: {e}")
except httpx.RequestError as e:
    print(f"Request error: {e}")

    说明：

  • HTTPStatusError：处理 4xx/5xx 错误。
  • RequestError：处理网络错误（如超时、连接失败）。
    • 3.7 HTTP/2 支持

    启用 HTTP/2 提高性能（需安装 h2）。

    import httpx

with httpx.Client(http2=True) as client:
    response = client.get("https://api.example.com")
    print(response.http_version)  # 输出: HTTP/2

    说明：

  • HTTP/2 提供多路复用和头部压缩，适合高并发请求。

    • 4. 性能与特点

  • 高效性：异步支持和连接池优化了高并发场景。
  • 现代设计：支持 HTTP/2 和 Brotli 压缩，性能优于 requests。
  • 兼容性：API 与 requests 类似，迁移成本低。
  • 灵活性：支持同步/异步、流式响应、自定义配置。
  • 局限性：
  • 异步代码需熟悉 asyncio 或 anyio。
  • HTTP/2 支持需额外安装 h2。
  • 生态集成：与 FastAPI、Starlette、Streamlit 等现代框架无缝协作。

    • 5. 实际应用场景

  • API 客户端：调用 RESTful API（如 GitHub、OpenAI）。
  • Web 爬虫：异步爬取网页或 API 数据。
  • 文件下载：流式下载大文件（如数据集、视频）。
  • 微服务交互：在异步 Web 应用中发送请求。
  • 测试工具：模拟 HTTP 请求测试后端服务。

    • 示例（异步 API 爬虫）：

    import httpx
import asyncio

async def fetch_url(client, url):
    response = await client.get(url)
    return response.json()

async def main():
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2",
        "https://api.example.com/data3"
    ]
    async with httpx.AsyncClient() as client:
        tasks = [fetch_url(client, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, result in zip(urls, results):
            print(f"{url}: {result}")

asyncio.run(main())

    说明：

  • 并发请求多个 API 端点，提高效率。

    • 6. 部署与 CLI

    httpx 提供命令行工具（需安装 httpx[cli]）。

    httpx https://api.example.com/data

    输出示例：

    {
  "key": "value"
}

    说明：

  • CLI 支持 GET、POST 等请求，适合快速测试 API。

    • 7. 注意事项

  • 同步 vs 异步：
  • 同步 API 适合简单脚本，异步 API 适合高并发。
  • 异步代码需在 asyncio.run 或事件循环中运行。
  • 错误处理：
  • 始终使用 raise_for_status() 检查 HTTP 错误。
  • 捕获 RequestError 处理网络问题。
  • 超时配置：
  • 默认超时可能过长，建议显式设置： 
    httpx.get("https://api.example.com", timeout=5.0)
  • HTTP/2：
  • 需服务器支持 HTTP/2，否则回退到 HTTP/1.1。
  • 安装 h2 确保 HTTP/2 可用。
  • 版本兼容性：
  • 最新版本（截至 2025，0.27.2）优化了性能和稳定性。
  • 检查 httpcore 和 h2 的版本兼容性。

    • 8. 综合示例

    以下是一个综合示例，展示同步/异步请求、流式下载和错误处理：

    import httpx
import asyncio
import os

# 同步请求
def sync_request():
    try:
        response = httpx.get(
            "https://api.github.com/repos/python/cpython",
            headers={"User-Agent": "MyApp/1.0"},
            timeout=5.0
        )
        response.raise_for_status()
        print("GitHub Repo:", response.json()["name"])
    except httpx.HTTPStatusError as e:
        print(f"HTTP error: {e}")
    except httpx.RequestError as e:
        print(f"Request error: {e}")

# 异步流式下载
async def download_file(url, filename):
    async with httpx.AsyncClient(http2=True) as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            with open(filename, "wb") as f:
                async for chunk in response.aiter_bytes():
                    f.write(chunk)
            print(f"Downloaded {filename}")

# 主函数
async def main():
    # 运行同步请求
    sync_request()

    # 异步下载文件
    await download_file(
        "https://example.com/sample.pdf",
        "sample.pdf"
    )

    # 并发请求多个 API
    urls = [
        "https://api.example.com/data1",
        "https://api.example.com/data2"
    ]
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        for url, resp in zip(urls, responses):
            if isinstance(resp, httpx.Response):
                print(f"{url}: {resp.status_code}")
            else:
                print(f"{url}: Error - {resp}")

if __name__ == "__main__":
    asyncio.run(main())

    说明：

  • 同步请求 GitHub API，获取仓库信息。
  • 异步下载文件，支持流式处理。
  • 并发请求多个 API，处理潜在错误。

    • 9. 资源与文档

  • 官方文档：https://www.python-httpx.org/
  • GitHub 仓库：https://github.com/encode/httpx
  • PyPI 页面：https://pypi.org/project/httpx/
  • 异步教程：https://www.python-httpx.org/async/
  • 与 requests 对比：https://www.python-httpx.org/compatibility/

    • 作者：彬彬侠

    物联沃分享整理
    物联沃-IOTWORD物联网 » 【Python】httpx库详解：HTTP客户端请求与响应处理指南

    代码收藏家 普通

    分享到：

    相关推荐

    发表回复