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/')
自定義頭部
要在傳入請求中包含其他標(biāo)頭,請使用headers關(guān)鍵字參數(shù):
header = {"user-agent": 'my_test/0001'} response = httpx.get("https://api.github.com/events", headers=header)
超時時間
httpx設(shè)置默認(rèn)的超時時間為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超時。如果想詳細(xì)設(shè)置,我們可以通過httpx.Timeout類來實(shí)現(xiàn):
# 讀取超時為10s,其他超時為5秒 timeout = httpx.Timeout(5, read=10) response = httpx.get('http://localhost:5000/update', timeout=timeout)
SSL證書
通過httpx發(fā)出HTTPS請求時,需要驗(yàn)證所請求主機(jī)的身份。我們可以通過verify來指定我們存在的CA證書:
response = httpx.get('https://example.org', verify='../../client.pem')
或者你可以傳遞標(biāo)準(zhǔn)庫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ù),請將2個元組得純文本str或bytes對象作為auth參數(shù)傳遞給請求函數(shù):
response = httpx.get('https://example.com', auth=('my_user', 'password123'))
要提供Digest Auth身份驗(yàn)證得憑據(jù),你需要Digest Auth使用純文本用戶名和密碼作為參數(shù)實(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ù)進(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ā)送一個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ù)
對于其他編碼,應(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", ...}]
二進(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)
對于大型下載,你可能需要使用不將整個響應(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請求重定向到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
如果你會使用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實(shí)例,那么你就可以通過調(diào)用.get()、.post()等方法發(fā)送請求。這些方法同樣支持timeout、auth、headers等參數(shù)來滿足我們的需求
合并/共享配置
httpx.Client還接收headers、cookie和params參數(shù),對于同一組的請求操作,將共享同樣的headers、cookie和params參數(shù)。如果請求方法中也包含了這些參數(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')
如此,這兩個請求的頭部都將包含{'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類實(shí)例,httpx.Limits類接收以下兩個參數(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)用程序。這對于兩個主要用例特別有用:
- 使用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í)例參數(shù)
- response: 響應(yīng)返回后調(diào)用。為一個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')
代理
要將所有請求使用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' }
對于上面匹配,如果未匹配到將不使用代理。域名后面還可以添加端口號,用于更加嚴(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ā)出異步請求,你需要一個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)容使用:
- 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, ...)
流式響應(yīng)
- Response.aread()
- Response.aiter_bytes()
- Response.aiter_text()
- Response.aiter_lines()
- Response.aiter_raw()
更多關(guān)于httpx的操作請看: https://www.python-httpx.org/
到此這篇關(guān)于python-httpx的具體使用的文章就介紹到這了,更多相關(guān)python httpx內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
sublime python3 輸入換行不結(jié)束的方法
下面小編就為大家分享一篇sublime python3 輸入換行不結(jié)束的方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2018-04-04淺談keras中的目標(biāo)函數(shù)和優(yōu)化函數(shù)MSE用法
這篇文章主要介紹了淺談keras中的目標(biāo)函數(shù)和優(yōu)化函數(shù)MSE用法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-06-06Python3標(biāo)準(zhǔn)庫之dbm UNIX鍵-值數(shù)據(jù)庫問題
dbm是面向DBM數(shù)據(jù)庫的一個前端,DBM數(shù)據(jù)庫使用簡單的字符串值作為鍵來訪問包含字符串的記錄。這篇文章主要介紹了Python3標(biāo)準(zhǔn)庫:dbm UNIX鍵-值數(shù)據(jù)庫的相關(guān)知識,需要的朋友可以參考下2020-03-03python神經(jīng)網(wǎng)絡(luò)學(xué)習(xí)數(shù)據(jù)增強(qiáng)及預(yù)處理示例詳解
這篇文章主要為大家介紹了python神經(jīng)網(wǎng)絡(luò)學(xué)習(xí)數(shù)據(jù)增強(qiáng)及預(yù)處理示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-05-05使用 Python 實(shí)現(xiàn)簡單的 switch/case 語句的方法
這篇文章主要介紹了用 Python 實(shí)現(xiàn)簡單的 switch/case 語句的方法,本文通過實(shí)例代碼給大家介紹的非常詳細(xì),具有一定的參考借鑒價值,需要的朋友可以參考下2018-09-09Python實(shí)現(xiàn)構(gòu)建一個儀表板的示例代碼
這篇文章主要為大家詳細(xì)介紹了Python如何實(shí)現(xiàn)構(gòu)建一個儀表板,文中的示例代碼講解詳細(xì),具有一定的參考價值,感興趣的小伙伴可以了解一下2023-03-03python torch.utils.data.DataLoader使用方法
這篇文章主要介紹了python torch.utils.data.DataLoader使用方法,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-04-04python函數(shù)運(yùn)行內(nèi)存時間等性能檢測工具
這篇文章主要為大家介紹了python函數(shù)運(yùn)行內(nèi)存時間等性能檢測工具,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-05-05