【PY模块】httpx 使用

介绍

用于 Python 的下一代 HTTP 客户端。

• 支持 同步 & 异步
• 支持 HTTP/1.1 和 HTTP/2
• 支持 HTTP & HTTPS 代理
• 还不支持 SOCKS 代理【aiohttp支持】
• 需要 Python 3.6+

安装

pip 安装

$ pip install httpx

HTTP/2 支持

pip install httpx[http2]

同步

跟 requests 基本一样

r = httpx.get('https://httpbin.org/get')
r
>>> <Response [200 OK]>

同步客户端（会话）

与使用顶级 API 相比，这可以带来显着的性能提升，包括：

• 减少请求之间的延迟（无握手）。
• 减少 CPU 使用率和往返次数。
• 减少网络拥塞。

还支持顶级 API 中不可用的功能，例如：

• 跨请求的 Cookie 持久性。
• 跨所有传出请求应用配置。
• 通过 HTTP 代理发送请求。
• 使用HTTP/2。

import httpx

# 上下文自动关闭会话
with httpx.Client() as client:
    res = client.get(url='https://bigdataboy.cn')
    print(res)
>>> <Response [200 OK]>

# 手动关闭会话
client = httpx.Client()
try:
    ...
finally:
    client.close()

异步

异步需要 httpx.AsyncClient()

import httpx
import asyncio

# 上下文方式
async def main():
    async with httpx.AsyncClient() as client:
        r = await client.get('https://bigdataboy.cn/')
        print(r)

# 普通方式 
async def main():
    client = httpx.AsyncClient()
    r = await client.get('https://bigdataboy.cn/')
    print(r)
    await client.aclose()

asyncio.run(main())
>>> <Response [200 OK]>

高级使用

同步普通写法，同步上下文写法，异步普通写法，异步上下文写法都是一样的用法，不一样的地方会标注

请求

httpx.get('https://bigdataboy.cn/', params=params)
# data: str & 表单 数据  json: json 数据 content: 字节数据
httpx.post('https://bigdataboy.cn/', data='', json={}, content=b'')
httpx.put('https://bigdataboy.cn/', data={'key': 'value'})
httpx.delete('https://bigdataboy.cn/')
httpx.head('https://bigdataboy.cn/')
httpx.options('https://bigdataboy.cn/')

响应

# 请求 url
r.url

# 文本显示
r.text

# Json 响应
r.json()

# 解码方式
r.encoding

# 设置解码方式
r.encoding = 'utf-8'

# 字节响应
r.content

# cookies
r.cookies
r.cookies[xxx]

# 响应状态码
r.status_code

# 响应 headers
r.headers

# 检查 http 版本
r.http_version

流式响应

# 二进制流式响应【同步模式】
with httpx.stream("GET", "https://www.example.com") as r:
     for data in r.iter_bytes():
         print(data)

# 文本流式响应【同步模式】
with httpx.stream("GET", "https://www.example.com") as r:
     for line in r.iter_lines():
         print(line)

# 二进制流式响应【异步模式】     
client = httpx.AsyncClient()
async with client.stream('GET', 'https://www.example.com/') as response:
     async for chunk in response.aiter_bytes():
     # response.aiter_text() 文本流式响应
     # response.aiter_raw() 用于流式传输原始响应字节，而不应用内容解码。

Cookie

cookies = {"k": "v"}
httpx.get('https://httpbin.org/cookies', cookies=cookies)

# 或者
cookies = httpx.Cookies()
cookies.set('cookie_on_domain', 'hello, there!', domain='httpbin.org')
cookies.set('cookie_off_domain', 'nope.', domain='example.org')
httpx.get('http://httpbin.org/cookies', cookies=cookies)

代理

还不支持 SOCKS 代理

# str 型
with httpx.Client(proxies="http://localhost:8030") as client:
    ...

# 字典型
proxies = {
    "http://": "http://localhost:8030",
    "https://": "http://localhost:8031",
}

with httpx.Client(proxies=proxies) as client:
    ...

重定向

httpx默认不会进行重定向跳转

# 不会重定向跳转
r = httpx.get('https://www.bigdataboy.cn/')
r.history # []
r.next_request # <Request('GET', 'https://bigdataboy.cn/')>

# 开启重定向跳转
r = httpx.get('https://www.bigdataboy.cn/', follow_redirects=True)
r.history # [<Response [301 Moved Permanently]>]
r.history[0] # <Response [301 Moved Permanently]>
r.history[0].url # https://www.bigdataboy.cn/
r.next_request # None

启用HTTP/2

需要服务端 支持 HTTP/2 才有用

client = httpx.AsyncClient(http2=True)

事件挂钩

httpx.AsyncClient() 需要异步 钩子函数

目前支持两种事件：

• request：在请求完全准备好之后，但在它被发送到网络之前调用。
• response：在从网络获取响应之后但在返回给调用者之前调用。

def log_request(request):
    print(f"Request event hook: {request.method} {request.url} - Waiting for response")

def log_response(response):
    request = response.request
    print(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")

client = httpx.Client(event_hooks={'request': [log_request], 'response': [log_response]})

事件允许对 request & response 进修改

def add_timestamp(request):
    request.headers['x-request-timestamp'] = datetime.now(tz=datetime.utc).isoformat()

client = httpx.Client(event_hooks={'request': [add_timestamp]})