Python-Requests庫進階用法——timeouts, retries, hooks
Python HTTP 請求庫在所有編程語言中是比較實用的程序。它簡單、直觀且在 Python 社區(qū)中無處不在。大多數(shù)與 HTTP 接口程序使用標準庫中的request或 urllib3。
由于簡單的API，請求很容易立即生效，但該庫還為高級需求提供了可擴展性。假如你正在編寫一個API密集型client或網(wǎng)路爬蟲，可能需要考慮網(wǎng)絡(luò)故障、靠譜的調(diào)試跟蹤和語法分析。
Request hooks
在使用第三方API時，通常需要驗證返回的響應(yīng)是否確實有效。Requests提供簡單有效的方法raise_for_status()，它斷言響應(yīng)HTTP狀態(tài)代碼不是4xx或5xx，即校驗請求沒有導(dǎo)致cclient或服務(wù)器錯誤。
比如：
response?=?requests.get('https://api.github.com/user/repos?page=1')
#?斷言沒有錯誤
response.raise_for_status()
如果每次調(diào)用都需要使用raise_for_status()，則此操作可能會重復(fù)。幸運的是，request庫提供了一個“hooks”(鉤子)接口，可以附加對請求過程某些部分的回調(diào)，確保從同一session對象發(fā)出的每個請求都會被檢查。
我們可以使用hooks來確保為每個響應(yīng)對象調(diào)用raise_for_status（）。
#?創(chuàng)建自定義請求對象時，修改全局模塊拋出錯誤異常
http?=?requests.Session()
assert_status_hook?=?lambda?response,?*args,?**kwargs:?response.raise_for_status()
http.hooks["response"]?=?[assert_status_hook]
http.get("https://api.github.com/user/repos?page=1")
>?HTTPError:?401?Client?Error:?Unauthorized?for?url:?https://api.github.com/user/repos?page=1
設(shè)置base URLs
requests中可以用兩種方法指定URL：
1、假設(shè)你只使用一個托管在API.org上的API，每次調(diào)用使用全部的URL地址
requests.get('https://api.org/list/')
requests.get('https://api.org/list/3/item')
2、安裝requests_toolbelt庫，使用BaseUrlSession指定base_url
from?requests_toolbelt?import?sessions
http?=?sessions.BaseUrlSession(base_url="https://api.org")
http.get("/list")
http.get("/list/item")
設(shè)置默認timeout值
Request官方文檔建議對所有的代碼設(shè)置超時。
如果你的python程序是同步的，忘記設(shè)置請求的默認timeout可能會導(dǎo)致你的請求或者有應(yīng)用程序掛起。
timeout的設(shè)定同樣有兩種方法：
1、每次都在get語句中指定timeout的值。
（不可取，只對本次請求有效）。
requests.get('https://github.com/',?timeout=0.001)
2、使用Transport Adapters設(shè)置統(tǒng)一的timeout時間（使用Transport Adapters，我們可以為所有HTTP調(diào)用設(shè)置默認超時，這確保了即使開發(fā)人員忘記在他的單個調(diào)用中添加timeout=1參數(shù)，也可以設(shè)置一個合理的超時，但這是允許在每個調(diào)用的基礎(chǔ)上重寫。）：
下面是一個帶有默認超時的自定義Transport Adapters的例子，在構(gòu)造http client和send()方法時，我們重寫構(gòu)造函數(shù)以提供默認timeout，以確保在沒有提供timeout參數(shù)時使用默認超時。
from?requests.adapters?import?HTTPAdapter
DEFAULT_TIMEOUT?=?5?#?seconds
class?TimeoutHTTPAdapter(HTTPAdapter):
????def?__init__(self,?*args,?**kwargs):
????????self.timeout?=?DEFAULT_TIMEOUT
????????if?"timeout"?in?kwargs:
????????????self.timeout?=?kwargs["timeout"]
????????????del?kwargs["timeout"]
????????super().__init__(*args,?**kwargs)
????def?send(self,?request,?**kwargs):
????????timeout?=?kwargs.get("timeout")
????????if?timeout?is?None:
????????????kwargs["timeout"]?=?self.timeout
????????return?super().send(request,?**kwargs)
實際代碼中，我們可以這樣使用：
import?requests
http?=?requests.Session()
#?此掛載對http和https都有效
adapter?=?TimeoutHTTPAdapter(timeout=2.5)
http.mount("https://",?adapter)
http.mount("http://",?adapter)
#?設(shè)置默認超時為2.5秒
response?=?http.get("https://api.twilio.com/")
#?通常為特定的請求重寫超時時間
response?=?http.get("https://api.twilio.com/",?timeout=10)
失敗時重試
網(wǎng)絡(luò)連接有丟包、擁擠，服務(wù)器出現(xiàn)故障。如果我們想要構(gòu)建一個真正健壯的程序，我們需要考慮失敗重試策略。
向HTTP client添加重試策略非常簡單。創(chuàng)建一個HTTPAdapter來適應(yīng)我們的策略。
from?requests.adapters?import?HTTPAdapter
from?requests.packages.urllib3.util.retry?import?Retry
retry_strategy?=?Retry(
????total=3,
????status_forcelist=[429,?500,?502,?503,?504],
????method_whitelist=["HEAD",?"GET",?"OPTIONS"]
)
adapter?=?HTTPAdapter(max_retries=retry_strategy)
http?=?requests.Session()
http.mount("https://",?adapter)
http.mount("http://",?adapter)
response?=?http.get("https://en.wikipedia.org/w/api.php")
其他參數(shù)：
最大重試次數(shù)
total=10引起重試的HTTP狀態(tài)碼
status_forcelist=[413, 429, 503]允許重試的請求方法
method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]兩次重試的間隔參數(shù)
backoff_factor=0
合并timeouts和retries--超時與重試
綜合上面學(xué)到的，我們可以通過這種方法將timeouts與retries結(jié)合到同一個Adapter中
retries?=?Retry(total=3,?backoff_factor=1,?status_forcelist=[429,?500,?502,?503,?504])
http.mount("https://",?TimeoutHTTPAdapter(max_retries=retries))
調(diào)試HTTP請求
如果一個HTTP請求失敗了，可以用下面兩種方法獲取失敗的信息：
使用內(nèi)置的調(diào)試日志
使用request hooks
打印HTTP頭部信息
將logging debug level設(shè)置為大于0的值都會將HTTP請求的頭部打印在日志中。當返回體過大或為字節(jié)流不便于日志時，打印頭部將非常有用。
import?requests
import?http
http.client.HTTPConnection.debuglevel?=?1
requests.get("https://www.google.com/")
#?Output?輸出信息
send:?b'GET?/?HTTP/1.1\r\nHost:?www.google.com\r\nUser-Agent:?python-requests/2.22.0\r\nAccept-Encoding:?gzip,?deflate\r\nAccept:?*/*\r\nConnection:?keep-alive\r\n\r\n'
reply:?'HTTP/1.1?200?OK\r\n'
header:?Date:?Fri,?28?Feb?2020?12:13:26?GMT
header:?Expires:?-1
header:?Cache-Control:?private,?max-age=0
打印所有HTTP內(nèi)容
當API返回內(nèi)容不太大時，我們可以使用request hooks與requests_toolbelt的dump工具庫輸出所有HTTP請求相應(yīng)內(nèi)容。
import?requests
from?requests_toolbelt.utils?import?dump
def?logging_hook(response,?*args,?**kwargs):
????data?=?dump.dump_all(response)
????print(data.decode('utf-8'))
http?=?requests.Session()
http.hooks["response"]?=?[logging_hook]
http.get("https://api.openaq.org/v1/cities",?params={"country":?"BA"})
# Output 輸出信息如下：
<?GET?/v1/cities?country=BA?HTTP/1.1
<?Host:?api.openaq.org
>?HTTP/1.1?200?OK
>?Content-Type:?application/json;?charset=utf-8
>?Transfer-Encoding:?chunked
>?Connection:?keep-alive
>
{
???"meta":{
??????"name":"openaq-api",
??????"license":"CC?BY?4.0",
??????"website":"https://docs.openaq.org/",
??????"page":1,
??????"limit":100,
??????"found":1
???},
???"results":[
??????{
?????????"country":"BA",
?????????"name":"Gora?de",
?????????"city":"Gora?de",
?????????"count":70797,
?????????"locations":1
??????}
???]
}
dump工具的用法：https://toolbelt.readthedocs.io/en/latest/dumputils.html
測試與模擬請求
測試第三方API有時不能一直發(fā)送真實的請求（比如按次收費的接口，還有沒開發(fā)完的=_=），測試中我們可以用getsentry/responses作為樁模塊攔截程序發(fā)出的請求并返回預(yù)定的數(shù)據(jù)，造成返回成功的假象。
class?TestAPI(unittest.TestCase):
[email protected]??#攔截該方法中的HTTP調(diào)用
????def?test_simple(self):
????????response_data?=?{
????????????????"id":?"ch_1GH8so2eZvKYlo2CSMeAfRqt",
????????????????"object":?"charge",
????????????????"customer":?{"id":?"cu_1GGwoc2eZvKYlo2CL2m31GRn",?"object":?"customer"},
????????????}
????????#?模擬?Stripe?API
????????responses.add(
????????????responses.GET,
????????????"https://api.stripe.com/v1/charges",
????????????json=response_data,
????????)
????????response?=?requests.get("https://api.stripe.com/v1/charges")
????????self.assertEqual(response.json(),?response_data)
一旦攔截成立就不能再向其他未設(shè)定過的URL發(fā)請求了，不然會報錯。
模仿瀏覽器行為
有些網(wǎng)頁會根據(jù)不同瀏覽器發(fā)送不同HTML代碼（為了反爬或適配設(shè)備），可以在發(fā)送請求時指定User-Agent將自己偽裝成特定瀏覽器。
import?requests
http?=?requests.Session()
http.headers.update({
????"User-Agent":?"Mozilla/5.0?(X11;?Ubuntu;?Linux?x86_64;?rv:68.0)?Gecko/20100101?Firefox/68.0"
})
總結(jié)：
以上就是Python-Requests庫的進階用法，在實際的代碼編寫中將會很有用，不管是開發(fā)編寫API還是測試在編寫自動化測試代碼，都會極大的提高所編寫代碼的穩(wěn)定性。