python-httpx的使用及說明

更新時(shí)間：2022年11月29日 08:51:43 作者：玉米叢里吃過虧

這篇文章主要介紹了python-httpx的使用及說明，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

安裝

pip install httpx

創(chuàng)建請(qǐng)求

通過httpx庫(kù)發(fā)出一個(gè)請(qǐng)求非常簡(jiǎn)單，如下：

import httpx

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

同樣，我們?cè)賮戆l(fā)出一個(gè)POST請(qǐng)求：

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

PUT, DELETE, HEAD和OPTIONS請(qǐng)求都遵循相同的樣式：

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/')

自定義頭部

要在傳入請(qǐng)求中包含其他標(biāo)頭，請(qǐng)使用headers關(guān)鍵字參數(shù)：

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

超時(shí)時(shí)間

httpx設(shè)置默認(rèn)的超時(shí)時(shí)間為5秒，超過此時(shí)間未響應(yīng)將引發(fā)錯(cuò)誤。

我們可以通過timeout關(guān)鍵字參數(shù)來手動(dòng)修改它：

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

你也可以將其設(shè)置為None完全禁用超時(shí)行為

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

超時(shí)又可以分為connect， read，write和pool超時(shí)。

如果想詳細(xì)設(shè)置，我們可以通過httpx.Timeout類來實(shí)現(xiàn)：

# 讀取超時(shí)為10s，其他超時(shí)為5秒
timeout = httpx.Timeout(5, read=10)
response = httpx.get('http://localhost:5000/update', timeout=timeout)

SSL證書

通過httpx發(fā)出HTTPS請(qǐng)求時(shí)，需要驗(yàn)證所請(qǐng)求主機(jī)的身份。我們可以通過verify來指定我們存在的CA證書：

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

或者你可以傳遞標(biāo)準(zhǔn)庫(kù)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設(shè)置為False禁用SSL驗(yàn)證：

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

認(rèn)證

HTTPX支持Basic Auth和Digest Auth身份驗(yàn)證。

要提供身份驗(yàn)證憑據(jù)，請(qǐng)將2個(gè)元組得純文本str或bytes對(duì)象作為auth參數(shù)傳遞給請(qǐng)求函數(shù)：

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

要提供Digest Auth身份驗(yàn)證得憑據(jù)，你需要Digest Auth使用純文本用戶名和密碼作為參數(shù)實(shí)例化一個(gè)對(duì)象。

然后可以將該對(duì)象作為auth參數(shù)傳遞給上述請(qǐng)求方法：

from httpx import DigestAuth

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

httpx還提供了一個(gè)FunctionAuth類，允許我們傳入一個(gè)Callable對(duì)象，該Callable接收request參數(shù)，并返回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)

請(qǐng)求類型

Query Params

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

此時(shí)我們打印一下URL，發(fā)現(xiàn)該URL已經(jīng)被正確編碼：

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

也可以傳遞一個(gè)列表數(shù)據(jù)進(jìn)去：

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

Form表單

通常情況下，你想要發(fā)送一些表單編碼數(shù)據(jù)，就像HTML表單一樣。

要做到這一點(diǎn)，你只需要將字典傳遞給data關(guān)鍵字參數(shù)即可：

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

如果你想要發(fā)送一個(gè)JSON數(shù)據(jù)，你可以通過將數(shù)據(jù)傳遞給json關(guān)鍵字參數(shù)即可：

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

二進(jìn)制數(shù)據(jù)

對(duì)于其他編碼，應(yīng)使用content關(guān)鍵字參數(shù)，傳遞一個(gè)bytes類型數(shù)據(jù)

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

響應(yīng)

響應(yīng)類型

在上面的栗子可以知道，我們每次請(qǐng)求之后都會(huì)返回一個(gè)httpx.Response對(duì)象，我們可以從此對(duì)象中獲取響應(yīng)內(nèi)容：

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

二進(jìn)制響應(yīng)

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

JSON響應(yīng)

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

流式響應(yīng)

對(duì)于大型下載，你可能需要使用不將整個(gè)響應(yīng)主體立即加載到內(nèi)存中的流式響應(yīng)。

你可以流式傳輸響應(yīng)的二進(jìn)制內(nèi)容：

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

流式傳輸響應(yīng)的文本內(nèi)容：

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

逐行流文本：

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

原始字節(jié)：

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

Cookie

如果響應(yīng)包含了Cookie，你可以這樣快速的訪問它：

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

重定向歷史

history響應(yīng)的屬性可用于檢查任何后續(xù)的重定向。

它包含遵循它們的順序的所有重定向響應(yīng)列表。

例如GitHub將所有HTTP請(qǐng)求重定向到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關(guān)鍵字參數(shù)來修改默認(rèn)得重定向處理：

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

httpx.Client

如果你會(huì)使用requests，那么可以使用httpx.Client代替requests.Session

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

另外，還可以使用.close()方法明確關(guān)閉連接池，而不會(huì)阻塞：

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

一旦你擁有了一個(gè)httpx.Client實(shí)例，那么你就可以通過調(diào)用.get()、.post()等方法發(fā)送請(qǐng)求。

這些方法同樣支持timeout、auth、headers等參數(shù)來滿足我們的需求

合并/共享配置

httpx.Client還接收headers、cookie和params參數(shù)，對(duì)于同一組的請(qǐng)求操作，將共享同樣的headers、cookie和params參數(shù)。

如果請(qǐng)求方法中也包含了這些參數(shù)，那么它們將進(jìn)行合并：

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')

如此，這兩個(gè)請(qǐng)求的頭部都將包含{'Token': '12345678'}。

請(qǐng)求1的params將會(huì)合并，請(qǐng)求2將會(huì)使用{'page_size': 1, 'size': 20}查詢參數(shù)

對(duì)于其他參數(shù)，如auth等，那么將會(huì)優(yōu)先使用請(qǐng)求方法里面的auth

base_url

httpx.Client還允許使用base_url參數(shù)來設(shè)置基礎(chǔ)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關(guān)鍵字參數(shù)來控制連接池的大小。它需要傳遞一個(gè)httpx.Limits類實(shí)例，httpx.Limits類接收以下兩個(gè)參數(shù)：

max_keepalive: 最大活躍連接數(shù)，設(shè)置為None表示無限制。默認(rèn)為10
max_connections：最大連接數(shù)，設(shè)置為None表示蘇限制。默認(rèn)為100

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

調(diào)用Python Web App

你可以配置httpx客戶端以使用WSGI協(xié)議直接調(diào)用Python Web應(yīng)用程序。這對(duì)于兩個(gè)主要用例特別有用：

使用httpx的測(cè)試案例中的客戶端
在測(cè)試期間或在dev/staging環(huán)境中模擬外部服務(wù)

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來使用給定的客戶端地址用于請(qǐng)求，如下：

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

如此，視圖函數(shù)home中request.remote_addr將會(huì)是1.2.3.4

事件鉤子

httpx允許你向客戶端注冊(cè)事件鉤子，每次發(fā)生特定類型的事件時(shí)都會(huì)調(diào)用該鉤子。httpx支持兩個(gè)事件鉤子：

request: 在即將發(fā)生請(qǐng)求時(shí)調(diào)用。為一個(gè)Callable列表，Callable接收httpx.Request實(shí)例參數(shù)
response: 響應(yīng)返回后調(diào)用。為一個(gè)Callable列表，Callable接收httpx.Response實(shí)例參數(shù)

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')

代理

要將所有請(qǐng)求使用http://localhost:8030的代理，請(qǐng)將代理URL傳遞給Client：

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

對(duì)于更高級(jí)的使用，請(qǐng)使用dict。例如，要將HTTP和HTTPS請(qǐng)求路由到兩個(gè)不同的代理: http://localhost:8030和http:localhost:8031:

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

代理所有請(qǐng)求：

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

代理域名為“example.com”的所有請(qǐng)求：

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

代理域名為“example.com”的所有HTTP請(qǐng)求：

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

代理所有包含“example.com”的所有請(qǐng)求：

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

對(duì)于上面匹配，如果未匹配到將不使用代理。

域名后面還可以添加端口號(hào)，用于更加嚴(yán)格的匹配。

此外，我們還可以將匹配設(shè)置為None，用于排除，如下：

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

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

異步支持

HTTPX默認(rèn)情況下提供標(biāo)準(zhǔn)的同步API，但是如果需要，還可以為你提供異步客戶端的選項(xiàng)。

要發(fā)出異步請(qǐng)求，你需要一個(gè)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()

發(fā)出請(qǐng)求

請(qǐng)求方法都是異步的，因此你應(yīng)該使用response = await client.get(...)樣式對(duì)以下所有內(nèi)容使用：