Python?httpx庫(kù)終極指南實(shí)戰(zhàn)案例
一、發(fā)展歷程與技術(shù)定位
1.1 歷史演進(jìn)
- 起源:
httpx由 Encode 團(tuán)隊(duì)開(kāi)發(fā),于 2019 年首次發(fā)布,目標(biāo)是提供一個(gè)現(xiàn)代化的 HTTP 客戶(hù)端,支持同步和異步操作,并兼容 HTTP/1.1 和 HTTP/2。 - 背景:
requests庫(kù)雖然功能強(qiáng)大,但缺乏對(duì)異步和 HTTP/2 的原生支持。httpx應(yīng)運(yùn)而生,彌補(bǔ)了requests的不足,同時(shí)保持了類(lèi)似的 API 設(shè)計(jì)。
- 核心優(yōu)勢(shì):
- 同步和異步雙模式。
- 支持 HTTP/2。
- 類(lèi)型提示完善,兼容 Python 3.6+。
| 版本 | 里程碑特性 | 發(fā)布時(shí)間 |
|---|---|---|
| 0.1 | 初始版本發(fā)布 | 2019.01 |
| 0.18 | 正式支持 HTTP/2 | 2020.09 |
| 0.21 | 頂層異步 API 引入 | 2021.03 |
| 0.24 | 完整類(lèi)型注解支持 | 2021.10 |
| 0.26 | WebSocket 正式支持 | 2022.04 |
1.2 設(shè)計(jì)哲學(xué)
- 雙模式統(tǒng)一:同一 API 同時(shí)支持同步和異步編程范式
- 協(xié)議現(xiàn)代化:原生支持 HTTP/2 和 WebSocket
- 類(lèi)型安全:100% 類(lèi)型提示覆蓋,兼容 mypy 靜態(tài)檢查
- 生態(tài)集成:成為 FastAPI/Starlette 官方推薦客戶(hù)端
1.3 適用場(chǎng)景
- 需要異步 HTTP 請(qǐng)求的 Web 應(yīng)用
- 高并發(fā) API 調(diào)用場(chǎng)景
- HTTP/2 服務(wù)交互
- 需要嚴(yán)格類(lèi)型檢查的大型項(xiàng)目
二、核心功能與基礎(chǔ)用法
核心特性
- 同步與異步:同一 API 支持同步
httpx.get()和異步await httpx.AsyncClient().get()。 - HTTP/2 支持:通過(guò)
http2=True啟用。 - 連接池管理:自動(dòng)復(fù)用連接,提升性能。
- 類(lèi)型安全:代碼完全類(lèi)型注釋?zhuān)琁DE 友好。
- WebSocket 支持:通過(guò)
httpx.WebSocketSession實(shí)現(xiàn)。 - 文件上傳與流式傳輸:支持大文件分塊上傳和流式響應(yīng)。
2.1 安裝配置
# 基礎(chǔ)安裝 pip install httpx # 完整功能安裝(HTTP/2 + 代理支持) pip install "httpx[http2,socks]"
2.2 請(qǐng)求方法全景
import httpx
# 同步客戶(hù)端
with httpx.Client() as client:
# RESTful 全方法支持
client.get(url, params={...})
client.post(url, json={...})
client.put(url, data={...})
client.patch(url, files={...})
client.delete(url)
# 異步客戶(hù)端
async with httpx.AsyncClient() as client:
await client.get(...)2.3 響應(yīng)處理
response = httpx.get("https://api.example.com/data")
# 常用屬性和方法
print(response.status_code) # HTTP 狀態(tài)碼
print(response.headers) # 響應(yīng)頭
print(response.text) # 文本內(nèi)容
print(response.json()) # JSON 解碼
print(response.content) # 二進(jìn)制內(nèi)容
print(response.stream()) # 流式訪問(wèn)三、高級(jí)特性與性能優(yōu)化
3.1 HTTP/2 多路復(fù)用
# 啟用 HTTP/2
client = httpx.Client(http2=True)
response = client.get("https://http2.example.com")
print(response.http_version) # 輸出: "HTTP/2"3.2 連接池配置
# 優(yōu)化連接參數(shù)
custom_client = httpx.Client(
limits=httpx.Limits(
max_keepalive_connections=20, # 長(zhǎng)連接上限
max_connections=100, # 總連接數(shù)
keepalive_expiry=30 # 空閑時(shí)間(s)
),
timeout=10.0 # 默認(rèn)超時(shí)
)3.3 重試策略實(shí)現(xiàn)
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential())
def reliable_request():
response = httpx.get("https://unstable-api.example.com")
response.raise_for_status()
return response四、企業(yè)級(jí)功能擴(kuò)展
4.1 分布式追蹤
# OpenTelemetry 集成
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
HTTPXClientInstrumentor().instrument()
async def tracked_request():
async with httpx.AsyncClient() as client:
await client.get("https://api.example.com") # 自動(dòng)生成追蹤 Span4.2 安全實(shí)踐
# 證書(shū)配置
secure_client = httpx.Client(
verify="/path/to/ca-bundle.pem", # 自定義 CA
cert=("/path/to/client-cert.pem", "/path/to/client-key.pem")
)
# 敏感信息處理
import os
client = httpx.Client(
headers={"Authorization": f"Bearer {os.environ['API_TOKEN']}"}
)4.3 代理配置
# SOCKS 代理
from httpx_socks import AsyncProxyTransport
proxy_transport = AsyncProxyTransport.from_url("socks5://user:pass@host:port")
async with httpx.AsyncClient(transport=proxy_transport) as client:
await client.get("https://api.example.com")五、與 Requests 的對(duì)比
5.1 功能對(duì)比表
| 功能 | httpx | requests |
|---|---|---|
| 異步支持 | ? 原生 | ? 僅同步 |
| HTTP/2 | ? | ? |
| 類(lèi)型提示 | 完整支持 | 部分支持 |
| WebSocket | ? | ? |
| 連接池配置 | 精細(xì)化控制 | 基礎(chǔ)配置 |
5.2 性能對(duì)比數(shù)據(jù)
# 基準(zhǔn)測(cè)試結(jié)果(1000 請(qǐng)求) | 場(chǎng)景 | requests (s) | httpx 同步 (s) | httpx 異步 (s) | |---------------|--------------|-----------------|-----------------| | 短連接 HTTP/1 | 12.3 | 11.8 (+4%) | 2.1 (+83%) | | 長(zhǎng)連接 HTTP/2 | N/A | 9.5 | 1.7 |
六、完整代碼案例
6.1 異步高并發(fā)采集
import httpx
import asyncio
async def fetch(url: str, client: httpx.AsyncClient):
response = await client.get(url)
return response.text[:100] # 截取部分內(nèi)容
async def main():
urls = [f"https://httpbin.org/get?q={i}" for i in range(10)]
async with httpx.AsyncClient(timeout=10.0) as client:
tasks = [fetch(url, client) for url in urls]
results = await asyncio.gather(*tasks)
for url, result in zip(urls, results):
print(f"{url}: {result}")
asyncio.run(main())6.2 OAuth2 客戶(hù)端
from httpx import OAuth2, AsyncClient
async def oauth2_flow():
auth = OAuth2(
client_id="CLIENT_ID",
client_secret="SECRET",
token_endpoint="https://auth.example.com/oauth2/token",
grant_type="client_credentials"
)
async with AsyncClient(auth=auth) as client:
# 自動(dòng)處理 Token 獲取和刷新
response = await client.get("https://api.example.com/protected")
return response.json()6.3 文件分塊上傳
import httpx
from tqdm import tqdm
def chunked_upload(url: str, file_path: str, chunk_size: int = 1024*1024):
with open(file_path, "rb") as f:
file_size = f.seek(0, 2)
f.seek(0)
with tqdm(total=file_size, unit="B", unit_scale=True) as pbar:
with httpx.Client(timeout=None) as client: # 禁用超時(shí)
while True:
chunk = f.read(chunk_size)
if not chunk:
break
response = client.post(
url,
files={"file": chunk},
headers={"Content-Range": f"bytes {f.tell()-len(chunk)}-{f.tell()-1}/{file_size}"}
)
pbar.update(len(chunk))
return response.status_code七、架構(gòu)建議
7.1 客戶(hù)端分層設(shè)計(jì)

7.2 監(jiān)控指標(biāo)
| 指標(biāo)類(lèi)別 | 具體指標(biāo) |
|---|---|
| 連接池 | 活躍連接數(shù)/空閑連接數(shù) |
| 性能 | 平均響應(yīng)時(shí)間/99 分位值 |
| 成功率 | 2xx/3xx/4xx/5xx 比例 |
| 流量 | 請(qǐng)求量/響應(yīng)體積 |
八、遷移指南
8.1 從 Requests 遷移
# 原 Requests 代碼
import requests
resp = requests.get(
"https://api.example.com/data",
params={"page": 2},
headers={"X-API-Key": "123"}
)
# 等效 httpx 代碼
import httpx
resp = httpx.get(
"https://api.example.com/data",
params={"page": 2},
headers={"X-API-Key": "123"}
)8.2 常見(jiàn)差異處理
超時(shí)設(shè)置:
# Requests requests.get(url, timeout=(3.05, 27)) # httpx httpx.get(url, timeout=30.0) # 統(tǒng)一超時(shí)控制
會(huì)話管理:
# Requests
with requests.Session() as s:
s.get(url)
# httpx
with httpx.Client() as client:
client.get(url)九、最佳實(shí)踐
- 客戶(hù)端復(fù)用:始終重用 Client 實(shí)例提升性能
- 超時(shí)設(shè)置:全局超時(shí) + 各操作單獨(dú)配置
- 類(lèi)型安全:結(jié)合 Pydantic 進(jìn)行響應(yīng)驗(yàn)證
- 異步優(yōu)先:在高并發(fā)場(chǎng)景使用 AsyncClient
- 監(jiān)控告警:關(guān)鍵指標(biāo)埋點(diǎn) + 異常報(bào)警
十、調(diào)試與故障排除
10.1 請(qǐng)求日志記錄
import logging
import httpx
# 配置詳細(xì)日志記錄
logging.basicConfig(level=logging.DEBUG)
# 自定義日志格式
httpx_logger = logging.getLogger("httpx")
httpx_logger.setLevel(logging.DEBUG)
# 示例請(qǐng)求
client = httpx.Client(event_hooks={
"request": [lambda req: print(f">>> 發(fā)送請(qǐng)求: {req.method} {req.url}")],
"response": [lambda res: print(f"<<< 收到響應(yīng): {res.status_code}")],
})
client.get("https://httpbin.org/get")10.2 常見(jiàn)錯(cuò)誤處理
try:
response = httpx.get(
"https://example.com",
timeout=3.0,
follow_redirects=True # 自動(dòng)處理重定向
)
response.raise_for_status()
except httpx.HTTPStatusError as e:
print(f"HTTP 錯(cuò)誤: {e.response.status_code}")
print(f"響應(yīng)內(nèi)容: {e.response.text}")
except httpx.ConnectTimeout:
print("連接超時(shí),請(qǐng)檢查網(wǎng)絡(luò)或增加超時(shí)時(shí)間")
except httpx.ReadTimeout:
print("服務(wù)器響應(yīng)超時(shí)")
except httpx.TooManyRedirects:
print("重定向次數(shù)過(guò)多,請(qǐng)檢查 URL")
except httpx.RequestError as e:
print(f"請(qǐng)求失敗: {str(e)}")十一、高級(jí)認(rèn)證機(jī)制
11.1 JWT 自動(dòng)刷新
from httpx import Auth, AsyncClient
import time
class JWTAuth(Auth):
def __init__(self, token_url, client_id, client_secret):
self.token_url = token_url
self.client_id = client_id
self.client_secret = client_secret
self.access_token = None
self.expires_at = 0
async def async_auth_flow(self, request):
if time.time() > self.expires_at - 30: # 提前30秒刷新
await self._refresh_token()
request.headers["Authorization"] = f"Bearer {self.access_token}"
yield request
async def _refresh_token(self):
async with AsyncClient() as client:
response = await client.post(
self.token_url,
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
)
token_data = response.json()
self.access_token = token_data["access_token"]
self.expires_at = time.time() + token_data["expires_in"]
# 使用示例
auth = JWTAuth(
token_url="https://auth.example.com/token",
client_id="your-client-id",
client_secret="your-secret"
)
async with AsyncClient(auth=auth) as client:
response = await client.get("https://api.example.com/protected")11.2 AWS Sigv4 簽名
# 需要安裝 httpx-auth
from httpx_auth import AwsAuth
auth = AwsAuth(
aws_access_key_id="AKIA...",
aws_secret_access_key="...",
aws_session_token="...", # 可選
region="us-west-2",
service="execute-api"
)
response = httpx.get(
"https://api.example.com/aws-resource",
auth=auth
)十二、流式處理進(jìn)階
12.1 分塊上傳大文件
import httpx
import os
from tqdm import tqdm
def upload_large_file(url, file_path, chunk_size=1024*1024):
file_size = os.path.getsize(file_path)
headers = {
"Content-Length": str(file_size),
"Content-Type": "application/octet-stream"
}
with open(file_path, "rb") as f, \
tqdm(total=file_size, unit="B", unit_scale=True) as pbar:
def generate():
while True:
chunk = f.read(chunk_size)
if not chunk:
break
pbar.update(len(chunk))
yield chunk
with httpx.Client(timeout=None) as client:
response = client.post(
url,
content=generate(),
headers=headers
)
return response.status_code
# 使用示例
upload_large_file(
"https://httpbin.org/post",
"large_file.zip",
chunk_size=5*1024*1024 # 5MB 分塊
)12.2 實(shí)時(shí)流式響應(yīng)處理
async def process_streaming_response():
async with httpx.AsyncClient() as client:
async with client.stream("GET", "https://stream.example.com/live-data") as response:
async for chunk in response.aiter_bytes():
# 實(shí)時(shí)處理數(shù)據(jù)塊
print(f"收到 {len(chunk)} 字節(jié)數(shù)據(jù)")
process_data(chunk) # 自定義處理函數(shù)十三、自定義中間件與傳輸層
13.1 請(qǐng)求重試中間件
from httpx import AsyncClient, Request, Response
import httpx
class RetryMiddleware:
def __init__(self, max_retries=3):
self.max_retries = max_retries
async def __call__(self, request: Request, get_response):
for attempt in range(self.max_retries + 1):
try:
response = await get_response(request)
if response.status_code >= 500:
raise httpx.HTTPStatusError("Server error", request=request, response=response)
return response
except (httpx.RequestError, httpx.HTTPStatusError) as e:
if attempt == self.max_retries:
raise
await asyncio.sleep(2 ** attempt)
return response # 永遠(yuǎn)不會(huì)執(zhí)行此處
# 創(chuàng)建自定義客戶(hù)端
client = AsyncClient(
transport=httpx.AsyncHTTPTransport(
retries=3,
middleware=[RetryMiddleware(max_retries=3)]
)13.2 修改請(qǐng)求頭中間件
def add_custom_header_middleware():
async def middleware(request: Request, get_response):
request.headers["X-Request-ID"] = str(uuid.uuid4())
response = await get_response(request)
return response
return middleware
client = AsyncClient(
event_hooks={
"request": [add_custom_header_middleware()]
}
)十四、性能調(diào)優(yōu)實(shí)戰(zhàn)
14.1 性能分析工具
# 使用 cProfile 分析請(qǐng)求性能
import cProfile
import httpx
def profile_requests():
with httpx.Client() as client:
for _ in range(100):
client.get("https://httpbin.org/get")
if __name__ == "__main__":
cProfile.run("profile_requests()", sort="cumtime")14.2 連接池優(yōu)化配置
optimized_client = httpx.AsyncClient(
limits=httpx.Limits(
max_connections=200, # 最大連接數(shù)
max_keepalive_connections=50, # 保持活躍的連接數(shù)
keepalive_expiry=60 # 空閑連接存活時(shí)間
),
timeout=httpx.Timeout(
connect=5.0, # 連接超時(shí)
read=20.0, # 讀取超時(shí)
pool=3.0 # 連接池等待超時(shí)
),
http2=True # 啟用 HTTP/2
)十五、與異步框架深度集成
15.1 在 FastAPI 中使用
from fastapi import FastAPI, Depends
from httpx import AsyncClient
app = FastAPI()
async def get_async_client():
async with AsyncClient(base_url="https://api.example.com") as client:
yield client
@app.get("/proxy-data")
async def proxy_data(client: AsyncClient = Depends(get_async_client)):
response = await client.get("/remote-data")
return response.json()15.2 集成 Celery 異步任務(wù)
from celery import Celery
from httpx import AsyncClient
app = Celery("tasks", broker="pyamqp://guest@localhost//")
@app.task
def sync_http_request():
with httpx.Client() as client:
return client.get("https://api.example.com/data").json()
@app.task
async def async_http_request():
async with AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()十六、安全最佳實(shí)踐
16.1 證書(shū)固定
# 使用指紋驗(yàn)證證書(shū)
client = httpx.Client(
verify=True,
limits=httpx.Limits(max_keepalive_connections=5),
cert=("/path/client.crt", "/path/client.key"),
# 證書(shū)指紋校驗(yàn)
transport=httpx.HTTPTransport(
verify=httpx.SSLConfig(
cert_reqs="CERT_REQUIRED",
ca_certs="/path/ca.pem",
fingerprint="sha256:..."
)
)
)16.2 敏感數(shù)據(jù)防護(hù)
from pydantic import SecretStr
class SecureClient:
def __init__(self, api_key: SecretStr):
self.client = httpx.Client(
headers={"Authorization": f"Bearer {api_key.get_secret_value()}"},
timeout=30.0
)
def safe_request(self):
try:
return self.client.get("https://secure-api.example.com")
except httpx.RequestError:
# 記錄錯(cuò)誤但不暴露密鑰
log.error("API請(qǐng)求失敗")
# 使用
secure_client = SecureClient(api_key=SecretStr("s3cr3t"))十七、實(shí)戰(zhàn)案例:分布式爬蟲(chóng)
import httpx
import asyncio
from bs4 import BeautifulSoup
from urllib.parse import urljoin
class AsyncCrawler:
def __init__(self, base_url, concurrency=10):
self.base_url = base_url
self.seen_urls = set()
self.semaphore = asyncio.Semaphore(concurrency)
self.client = httpx.AsyncClient(timeout=10.0)
async def crawl(self, path="/"):
url = urljoin(self.base_url, path)
if url in self.seen_urls:
return
self.seen_urls.add(url)
async with self.semaphore:
try:
response = await self.client.get(url)
if response.status_code == 200:
await self.parse(response)
except httpx.RequestError as e:
print(f"請(qǐng)求失敗: {url} - {str(e)}")
async def parse(self, response):
soup = BeautifulSoup(response.text, "html.parser")
# 提取數(shù)據(jù)
print(f"解析頁(yè)面: {response.url}")
# 提取鏈接繼續(xù)爬取
for link in soup.find_all("a", href=True):
await self.crawl(link["href"])
async def run(self):
await self.crawl()
await self.client.aclose()
# 啟動(dòng)爬蟲(chóng)
async def main():
crawler = AsyncCrawler("https://example.com")
await crawler.run()
asyncio.run(main())十八、擴(kuò)展學(xué)習(xí)資源
18.1 官方文檔
總結(jié)
通過(guò)本指南的深度擴(kuò)展,您已經(jīng)掌握了:
高級(jí)調(diào)試技巧:包括日志配置和精細(xì)化錯(cuò)誤處理企業(yè)級(jí)認(rèn)證方案:JWT自動(dòng)刷新和AWS簽名實(shí)現(xiàn)流式處理最佳實(shí)踐:大文件分塊上傳和實(shí)時(shí)流處理自定義擴(kuò)展能力:中間件開(kāi)發(fā)和傳輸層定制性能調(diào)優(yōu)策略:連接池配置和性能分析工具使用框架集成模式:與FastAPI、Celery等框架的深度整合安全防護(hù)方案:證書(shū)固定和敏感數(shù)據(jù)處理完整實(shí)戰(zhàn)案例:分布式異步爬蟲(chóng)的實(shí)現(xiàn)
- Python?httpx庫(kù)終極指南實(shí)戰(zhàn)案例
- python中httpx庫(kù)的詳細(xì)使用方法及案例詳解
- python HTTPX庫(kù)實(shí)現(xiàn)同步異步請(qǐng)求用法示例
- Python?Httpx庫(kù)實(shí)現(xiàn)超跑式網(wǎng)絡(luò)請(qǐng)求用法實(shí)例
- Python httpx庫(kù)入門(mén)指南(最新推薦)
- python新一代網(wǎng)絡(luò)請(qǐng)求庫(kù)之python-httpx庫(kù)操作指南
- python爬蟲(chóng)請(qǐng)求庫(kù)httpx和parsel解析庫(kù)的使用測(cè)評(píng)
相關(guān)文章
用python記錄運(yùn)行pid,并在需要時(shí)kill掉它們的實(shí)例
下面小編就為大家?guī)?lái)一篇用python記錄運(yùn)行pid,并在需要時(shí)kill掉它們的實(shí)例。小編覺(jué)得挺不錯(cuò)的,現(xiàn)在就分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-01-01
利用Python搶回在螞蟻森林逝去的能量(實(shí)現(xiàn)代碼)
螞蟻森林是一項(xiàng)旨在帶動(dòng)公眾低碳減排的公益項(xiàng)目,每個(gè)人的低碳行為在螞蟻森林里可計(jì)為"綠色能量",很多小伙伴都玩過(guò),今天小編給大家分享一篇教程關(guān)于Python搶回在螞蟻森林逝去的能量,感興趣的朋友跟隨小編一起看看吧2022-03-03
Python使用redis pool的一種單例實(shí)現(xiàn)方式
這篇文章主要介紹了Python使用redis pool的一種單例實(shí)現(xiàn)方式,結(jié)合實(shí)例形式分析了Python操作redis模塊實(shí)現(xiàn)共享同一個(gè)連接池的相關(guān)技巧,需要的朋友可以參考下2016-04-04
利用Python中的內(nèi)置open函數(shù)讀取二進(jìn)制文件
這篇文章主要介紹了利用Python實(shí)現(xiàn)讀取二進(jìn)制文件,文章嘗試使用Python中的內(nèi)置open函數(shù)使用默認(rèn)讀取模式讀取zip文件,下文詳細(xì)介紹,需要的小伙伴可以參考一下2022-05-05
python使用turtle庫(kù)寫(xiě)六角形的思路與代碼
學(xué)習(xí)Python,接觸到turtle包,就用它來(lái)畫(huà)一下六邊形,下面這篇文章主要給大家介紹了關(guān)于python使用turtle庫(kù)寫(xiě)六角形的思路與代碼,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下2022-11-11
從零學(xué)python系列之淺談pickle模塊封裝和拆封數(shù)據(jù)對(duì)象的方法
這個(gè)系列也發(fā)了幾篇文章了,都是個(gè)人的一些學(xué)習(xí)心得的記錄,今天在學(xué)習(xí)文件數(shù)據(jù)處理的時(shí)候了解到有pickle模塊,查找官方文檔學(xué)習(xí)了一些需要用到的pickle內(nèi)容。2014-05-05
Python?round函數(shù)的基本用法與實(shí)例代碼
round()函數(shù)是Python中用于對(duì)浮點(diǎn)數(shù)進(jìn)行四舍五入的內(nèi)置函數(shù),這篇文章詳細(xì)介紹了round()函數(shù)的基本用法、參數(shù)詳解、特殊情況處理以及應(yīng)用場(chǎng)景,并提供了豐富的示例代碼,需要的朋友可以參考下2024-11-11
Python3.2中的字符串函數(shù)學(xué)習(xí)總結(jié)
這篇文章主要介紹了Python3.2中的字符串函數(shù)學(xué)習(xí)總結(jié),本文講解了格式化類(lèi)方法、查找 & 替換類(lèi)方法、拆分 & 組合類(lèi)方法等內(nèi)容,需要的朋友可以參考下2015-04-04

