python-httpx的具體使用

Posted on 2022-01-09 by WalkonNet

安裝

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/

到此這篇關於python-httpx的具體使用的文章就介紹到這瞭,更多相關python httpx內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

python-httpx的具體使用

目錄

安裝

創建請求

自定義頭部

超時時間

SSL證書

認證

請求類型

Query Params

Form表單

文件上傳

JSON

二進制數據

響應

響應類型

Cookie

重定向歷史

httpx.Client

合並/共享配置

base_url

limits

調用Python Web App

事件鉤子

代理

異步支持

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

安裝

創建請求

自定義頭部

超時時間

SSL證書

認證

請求類型

Query Params

Form表單

文件上傳

JSON

二進制數據

響應

響應類型

Cookie

重定向歷史

httpx.Client

合並/共享配置

base_url

limits

調用Python Web App

事件鉤子

代理

異步支持

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆