python-httpx的具体使用

目录

• 安装
• 创建请求
• 自定义头部
• 超时时间
• SSL证书
• 认证
• 请求类型
• Query Params
• Form表单
• 文件上传
• JSON
• 二进制数据
• 响应
• 响应类型
• Cookie
• 重定向历史
• httpx.Client
• 合并/共享配置
• base_url
• limits
• 调用Python Web App
• 事件钩子
• 代理
• 异步支持

HTTPX是Python3的功能齐全的HTTP客户端，它提供同步和异步API，并支持HTTP/1.1和HTTP/2

安装

pip install httpx

创建请求

通过httpx库发出一个请求非常简单，如下：

import httpx

response = httpx.get(\'https://www.baidu.com/\')
print(type(response), response)     # <class \'httpx.Response\'> <Response [200 OK]>

同样，我们再来发出一个POST请求：

response = httpx.post(\'http://localhost:5000/login\', data={\'username\': \'httpx\', \'password\': \'123456\'})

PUT, DELETE, HEAD和OPTIONS请求都遵循相同的样式：

response = httpx.put(\'http://www.baidu.com/\', data={key: value})
response = httpx.head(\'http://www.baidu.com/\')
response = httpx.delete(\'http://www.baidu.com/\')
response = httpx.options(\'http://www.baidu.com/\')

自定义头部

要在传入请求中包含其他标头，请使用headers关键字参数：

header = {\"user-agent\": \'my_test/0001\'}
response = httpx.get(\"https://api.github.com/events\", headers=header)

超时时间

httpx设置默认的超时时间为5秒，超过此时间未响应将引发错误。我们可以通过timeout关键字参数来手动修改它：

response = httpx.get(\'http://localhost:5000/update\', timeout=10)

你也可以将其设置为None完全禁用超时行为

response = httpx.get(\'http://localhost:5000/update\', timeout=None)

超时又可以分为connect， read，write和pool超时。如果想详细设置，我们可以通过httpx.Timeout类来实现：

# 读取超时为10s，其他超时为5秒
timeout = httpx.Timeout(5, read=10)
response = httpx.get(\'http://localhost:5000/update\', timeout=timeout)

SSL证书

通过httpx发出HTTPS请求时，需要验证所请求主机的身份。我们可以通过verify来指定我们存在的CA证书：

response = httpx.get(\'https://example.org\', verify=\'../../client.pem\')

或者你可以传递标准库ssl.SSLContext

import ssl
import httpx

context = ssl.create_default_context()
context.load_verify_locations(cafile=\'../../client.pem\')
response = httpx.get(\'https://example.org\', verify=\'../../client.pem\')

又或者，你可以将verify设置为False禁用SSL验证：

response = httpx.get(\'https://example.org\', verify=False)

认证

HTTPX支持Basic Auth和Digest Auth身份验证。要提供身份验证凭据，请将2个元组得纯文本str或bytes对象作为auth参数传递给请求函数：

response = httpx.get(\'https://example.com\', auth=(\'my_user\', \'password123\'))

要提供Digest Auth身份验证得凭据，你需要Digest Auth使用纯文本用户名和密码作为参数实例化一个对象。然后可以将该对象作为auth参数传递给上述请求方法：

from httpx import DigestAuth

auth = DigestAuth(\'my_user\', \'password123\')
response = httpx.get(\'https://example.com\', auth=auth)

httpx还提供了一个FunctionAuth类，允许我们传入一个Callable对象，该Callable接收request参数，并返回request。如下：

import httpx
from httpx._auth import FunctionAuth

def init_authorization(request):
    request.headers[\'Authorization\'] = \'Bearer 12334\'
    yield request

auth = FunctionAuth(init_authorization)
response = httpx.get(\'http://localhost:5000/home\', auth=auth)

请求类型

Query Params

params = {\"name\":\"zhangsan\", \"age\":18}
response = httpx.get(\"https://www.baidu.com/s\", params=params)

此时我们打印一下URL，发现该URL已经被正确编码：

print(response.url)    # https://www.baidu.com/s?name=zhangsan&age=18

也可以传递一个列表数据进去：

params = {\"name\":\"zhangsan\", \"favorite\": [\"football\", \"basketball\"]}
response = httpx.get(\"https://www.baidu.com/s\", params=params)

Form表单

通常情况下，你想要发送一些表单编码数据，就像HTML表单一样。要做到这一点，你只需要将字典传递给data关键字参数即可：

data = {\'name\': \'张三\'}
response = httpx.post(\'http://127.0.0.1:5000/test/post\', data=data)

文件上传

你还可以使用HTTP分段编码上传文件

f = open(\'a.txt\', \'rb\')
files = {\'file\': f}
response = httpx.post(\'http://localhost:5000/post\', files=files)
f.close()

JSON

如果你想要发送一个JSON数据，你可以通过将数据传递给json关键字参数即可：

response = httpx.post(\'http://127.0.0.1:5000/test/post\', json={\'name\': \'张三\'})

二进制数据

对于其他编码，应使用content关键字参数，传递一个bytes类型数据

content = b\'Hello World\'
response = httpx.post(\'http://127.0.0.1:5000/test/post\', content=content)

响应

响应类型

在上面的栗子可以知道，我们每次请求之后都会返回一个httpx.Response对象，我们可以从此对象中获取响应内容：

response = httpx.get(\"https://api.github.com/events\")
print(type(response.text), response.text)      # <class \'str\'> [{\"id\":\"14551634865\",\"type\":\"PushEvent\", ...}]

二进制响应

print(type(response.content), response.content) # <class \'bytes\'> b\'[{\"id\":\"14551634865\",\"type\":\"PushEvent\", ...}]

JSON响应

print(type(response.json()), response.json())   # <class \'list\'> [{\'id\': \'14551634865\', \'type\': \'PushEvent\', ...}]

流式响应

对于大型下载，你可能需要使用不将整个响应主体立即加载到内存中的流式响应。你可以流式传输响应的二进制内容：

for data in response.iter_bytes():
    print(data)

流式传输响应的文本内容：

for text in response.iter_text():
    print(text)

逐行流文本：

for line in response.iter_lines():
    print(line)

原始字节：

for chunk in response.iter_raw():
    print(chunk)

Cookie

如果响应包含了Cookie，你可以这样快速的访问它：

response = httpx.get(\'http://localhost:5050/get\')
print(response.cookies[\'user\'])

重定向历史

history响应的属性可用于检查任何后续的重定向。它包含遵循它们的顺序的所有重定向响应列表。例如GitHub将所有HTTP请求重定向到HTTPS:

response = httpx.get(\'http://github.com/\')
print(response, response.url)       # <Response [200 OK]> https://github.com/
print(response.history, response.history[0].url)    # [<Response [301 Moved Permanently]>] http://github.com/

你还可以使用allow_redirects关键字参数来修改默认得重定向处理：

response = httpx.get(\'http://github.com/\', allow_redirects=False)
print(response)         # <Response [301 Moved Permanently]>
print(response.history) # []

httpx.Client

如果你会使用requests，那么可以使用httpx.Client代替requests.Session

with httpx.Client() as client:
    response = client.get(\'http://localhost:5000/details\')

另外，还可以使用.close()方法明确关闭连接池，而不会阻塞：

client = httpx.Client()
try:
    response = client.get(\'http://localhost:5000/details\')
finally:
    client.close()

一旦你拥有了一个httpx.Client实例，那么你就可以通过调用.get()、.post()等方法发送请求。这些方法同样支持timeout、auth、headers等参数来满足我们的需求

合并/共享配置

httpx.Client还接收headers、cookie和params参数，对于同一组的请求操作，将共享同样的headers、cookie和params参数。如果请求方法中也包含了这些参数，那么它们将进行合并：

with httpx.Client(headers={\'Token\': \'12345678\'}, params={\'page_size\': 1, \'size\': 20}) as client:
    resp1 = client.get(\'http://localhost:5000/get\', params={\'search\': \'laozhang\'})
    resp2 = client.post(\'http://localhost:5000/post\')

如此，这两个请求的头部都将包含{'Token': '12345678'}。请求1的params将会合并，请求2将会使用{'page_size': 1, 'size': 20}查询参数

对于其他参数，如auth等，那么将会优先使用请求方法里面的auth

base_url

httpx.Client还允许使用base_url参数来设置基础URL，如下：

with httpx.Client(base_url=\'http://localhost:5000\') as client:
    response = client.get(\'/user/detail\')
    print(response.url)     # http://localhost:5050/user/detail

limits

可以使用limits关键字参数来控制连接池的大小。它需要传递一个httpx.Limits类实例，httpx.Limits类接收以下两个参数：

• max_keepalive: 最大活跃连接数，设置为None表示无限制。默认为10
• max_connections：最大连接数，设置为None表示苏限制。默认为100

limits = httpx.Limits(max_keepalive=2, max_connections=5)
client = httpx.Client(limits=limits)

调用Python Web App

你可以配置httpx客户端以使用WSGI协议直接调用Python Web应用程序。这对于两个主要用例特别有用：

• 使用httpx的测试案例中的客户端
• 在测试期间或在dev/staging环境中模拟外部服务

import httpx
from flask import Flask

app = Flask(__name__)

@app.route(\"/home\")
def home():
    return \'Home Api Success\'

with httpx.Client(app=app, base_url=\'http://testapi\') as client:
    response = client.get(\'/home\')
    print(response)                     # <Response [200 OK]>
    print(response.text, response.url)  # Home Api Success http://testapi/home

我们还可以通过使用WSGITransport来使用给定的客户端地址用于请求，如下：

transport = httpx.WSGITransport(app=app, remote_addr=\'1.2.3.4\')
with httpx.Client(transport=transport, base_url=\'http://testapi\') as client:
    response = client.get(\'/home\')
    print(response)                     # <Response [200 OK]>
    print(response.text, response.url)  # Home Api Success http://testapi/home

如此，视图函数home中request.remote_addr将会是1.2.3.4

事件钩子

httpx允许你向客户端注册事件钩子，每次发生特定类型的事件时都会调用该钩子。httpx支持两个事件钩子：

• request: 在即将发生请求时调用。为一个Callable列表，Callable接收httpx.Request实例参数
• response: 响应返回后调用。为一个Callable列表，Callable接收httpx.Response实例参数

def _log_request(request):
    print(type(request), request.url, request.method)       # <class \'httpx.Request\'> http://localhost:5000/hello GET
    
def _log_response(response):
    print(type(response), response.url, response.text)      # <class \'httpx.Response\'> http://localhost:5000/hello Home Api Success
    
with httpx.Client(base_url=\'http://localhost:5000\') as client:
    client.get(\'/home\')

代理

要将所有请求使用http://localhost:8030的代理，请将代理URL传递给Client：

with httpx.Client(proxies=\'http://localhost:8030\') as client:
    pass

对于更高级的使用，请使用dict。例如，要将HTTP和HTTPS请求路由到两个不同的代理: http://localhost:8030和http:localhost:8031:

proxies = {
    \'http://\': \'http://localhost:8030\',
    \'https://\': \'http:localhost:8031\'
}
with httpx.Client(proxies=proxies) as client:
    pass

代理所有请求：

proxies = {
    \'all://\': \'http://localhost:8030\'
}

代理域名为“example.com”的所有请求：

proxies = {
    \'all://example.com\': \'http://localhost:8030\'
}

代理域名为“example.com”的所有HTTP请求：

proxies = {
    \'http://example.com\': \'http://localhost:8030\'
}

代理所有包含“example.com”的所有请求：

proxies = {
    \'all://*example.com\': \'http://localhost:8030\'
}

对于上面匹配，如果未匹配到将不使用代理。域名后面还可以添加端口号，用于更加严格的匹配。此外，我们还可以将匹配设置为None，用于排除，如下：

proxies = {
    \'all://\': \'http://localhost:8030\',
    \'all://example.com\': None
}

即除使用“example.com”域名的路由，将使用“http://localhost:8030”代理

异步支持

HTTPX默认情况下提供标准的同步API，但是如果需要，还可以为你提供异步客户端的选项。要发出异步请求，你需要一个httpx.AsyncClient

import asyncio
import httpx

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get(\'https://example.org/\')

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

发出请求

请求方法都是异步的，因此你应该使用response = await client.get(…)样式对以下所有内容使用：

• AsyncClient.get(url, …)
• AsyncClient.options(url, …)
• AsyncClient.head(url, …)
• AsyncClient.post(url, …)
• AsyncClient.put(url, …)
• AsyncClient.patch(url, …)
• AsyncClient.delete(url, …)
• AsyncClient.request(url, …)
• AsyncClient.send(url, …)

流式响应

• Response.aread()
• Response.aiter_bytes()
• Response.aiter_text()
• Response.aiter_lines()
• Response.aiter_raw()

更多关于httpx的操作请看: https://www.python-httpx.org/