python-httpx的使用及說明

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

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

安裝

pip install httpx

創(chuàng)建請求

通過httpx庫發(fā)出一個請求非常簡單，如下：

import httpx

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

同樣，我們再來發(fā)出一個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關(guān)鍵字參數(shù)：

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

超時時間

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

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

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

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

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

超時又可以分為connect， read，write和pool超時。

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

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

SSL證書

通過httpx發(fā)出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設(shè)置為False禁用SSL驗證：

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

認證

HTTPX支持Basic Auth和Digest Auth身份驗證。

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

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

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

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

from httpx import DigestAuth

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

httpx還提供了一個FunctionAuth類，允許我們傳入一個Callable對象，該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)

請求類型

Query Params

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

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

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

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

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

Form表單

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

要做到這一點，你只需要將字典傳遞給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ā)送一個JSON數(shù)據(jù)，你可以通過將數(shù)據(jù)傳遞給json關(guān)鍵字參數(shù)即可：

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

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

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

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

響應(yīng)

響應(yīng)類型

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

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

二進制響應(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)

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

你可以流式傳輸響應(yīng)的二進制內(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請求重定向到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ù)來修改默認得重定向處理：

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()方法明確關(guān)閉連接池，而不會阻塞：

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

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

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

合并/共享配置

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

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

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}查詢參數(shù)

對于其他參數(shù)，如auth等，那么將會優(yōu)先使用請求方法里面的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ù)來控制連接池的大小。它需要傳遞一個httpx.Limits類實例，httpx.Limits類接收以下兩個參數(shù)：

max_keepalive: 最大活躍連接數(shù)，設(shè)置為None表示無限制。默認為10
max_connections：最大連接數(shù)，設(shè)置為None表示蘇限制。默認為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)用程序。這對于兩個主要用例特別有用：

使用httpx的測試案例中的客戶端
在測試期間或在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來使用給定的客戶端地址用于請求，如下：

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將會是1.2.3.4

事件鉤子

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

request: 在即將發(fā)生請求時調(diào)用。為一個Callable列表，Callable接收httpx.Request實例參數(shù)
response: 響應(yīng)返回后調(diào)用。為一個Callable列表，Callable接收httpx.Response實例參數(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')

代理

要將所有請求使用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'
}

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

域名后面還可以添加端口號，用于更加嚴格的匹配。

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

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

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

異步支持

HTTPX默認情況下提供標準的同步API，但是如果需要，還可以為你提供異步客戶端的選項。

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

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