python-httpx的使用及說明
HTTPX是Python3的功能齊全的HTTP客戶端,它提供同步和異步API,并支持HTTP/1.1和HTTP/2
安裝
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關鍵字參數(shù):
header = {"user-agent": 'my_test/0001'} response = httpx.get("https://api.github.com/events", headers=header)
超時時間
httpx
設置默認的超時時間為5秒,超過此時間未響應將引發(fā)錯誤。
我們可以通過timeout
關鍵字參數(shù)來手動修改它:
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
類來實現(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
設置為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
關鍵字參數(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
關鍵字參數(shù)即可:
response = httpx.post('http://127.0.0.1:5000/test/post', json={'name': '張三'})
二進制數(shù)據(jù)
對于其他編碼,應使用content
關鍵字參數(shù),傳遞一個bytes
類型數(shù)據(jù)
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)
原始字節(jié):
for chunk in response.iter_raw(): ? ? print(chunk)
Cookie
如果響應包含了Cookie,你可以這樣快速的訪問它:
response = httpx.get('http://localhost:5050/get') print(response.cookies['user'])
重定向歷史
history
響應的屬性可用于檢查任何后續(xù)的重定向。
它包含遵循它們的順序的所有重定向響應列表。
例如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
關鍵字參數(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()
方法明確關閉連接池,而不會阻塞:
client = httpx.Client() try: response = client.get('http://localhost:5000/details') finally: client.close()
一旦你擁有了一個httpx.Client
實例,那么你就可以通過調用.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ù)來設置基礎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
關鍵字參數(shù)來控制連接池的大小。它需要傳遞一個httpx.Limits
類實例,httpx.Limits
類接收以下兩個參數(shù):
- max_keepalive: 最大活躍連接數(shù),設置為None表示無限制。默認為10
- max_connections:最大連接數(shù),設置為None表示蘇限制。默認為100
limits = httpx.Limits(max_keepalive=2, max_connections=5) client = httpx.Client(limits=limits)
調用Python Web App
你可以配置httpx
客戶端以使用WSGI協(xié)議直接調用Python Web應用程序。這對于兩個主要用例特別有用:
- 使用
httpx
的測試案例中的客戶端 - 在測試期間或在dev/staging環(huán)境中模擬外部服務
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ā)生特定類型的事件時都會調用該鉤子。httpx
支持兩個事件鉤子:
request
: 在即將發(fā)生請求時調用。為一個Callable列表,Callable接收httpx.Request
實例參數(shù)response
: 響應返回后調用。為一個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' }
對于上面匹配,如果未匹配到將不使用代理。
域名后面還可以添加端口號,用于更加嚴格的匹配。
此外,我們還可以將匹配設置為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ā)出請求
請求方法都是異步的,因此你應該使用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/
以上為個人經(jīng)驗,希望能給大家一個參考,也希望大家多多支持腳本之家。