相信我，這么寫(xiě)Python代碼，老板給你漲工資
圖片來(lái)自愛(ài)美劇
Python HTTP 請(qǐng)求庫(kù)在所有編程語(yǔ)言中是比較實(shí)用的程序。它簡(jiǎn)單、直觀且在 Python 社區(qū)中無(wú)處不在。大多數(shù)與 HTTP 接口程序使用標(biāo)準(zhǔn)庫(kù)中的request或 urllib3。
由于簡(jiǎn)單的API，請(qǐng)求很容易立即生效，但該庫(kù)還為高級(jí)需求提供了可擴(kuò)展性。假如你正在編寫(xiě)一個(gè)API密集型client或網(wǎng)路爬蟲(chóng)，可能需要考慮網(wǎng)絡(luò)故障、靠譜的調(diào)試跟蹤和語(yǔ)法分析。
Request hooks
在使用第三方API時(shí)，通常需要驗(yàn)證返回的響應(yīng)是否確實(shí)有效。Requests提供簡(jiǎn)單有效的方法raise_for_status()，它斷言響應(yīng)HTTP狀態(tài)代碼不是4xx或5xx，即校驗(yàn)請(qǐng)求沒(méi)有導(dǎo)致cclient或服務(wù)器錯(cuò)誤。
比如：
response = requests.get('https://api.github.com/user/repos?page=1')
# 斷言沒(méi)有錯(cuò)誤
response.raise_for_status()
如果每次調(diào)用都需要使用raise_for_status()，則此操作可能會(huì)重復(fù)。幸運(yùn)的是，request庫(kù)提供了一個(gè)“hooks”(鉤子)接口，可以附加對(duì)請(qǐng)求過(guò)程某些部分的回調(diào)，確保從同一session對(duì)象發(fā)出的每個(gè)請(qǐng)求都會(huì)被檢查。
我們可以使用hooks來(lái)確保為每個(gè)響應(yīng)對(duì)象調(diào)用raise_for_status（）。
# 創(chuàng)建自定義請(qǐng)求對(duì)象時(shí)，修改全局模塊拋出錯(cuò)誤異常
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è)你只使用一個(gè)托管在API.org上的API，每次調(diào)用使用全部的URL地址
requests.get('https://api.org/list/')
requests.get('https://api.org/list/3/item')
2、安裝requests_toolbelt庫(kù)，使用BaseUrlSession指定base_url
from requests_toolbelt import sessions
http = sessions.BaseUrlSession(base_url="https://api.org")
http.get("/list")
http.get("/list/item")
設(shè)置默認(rèn)timeout值
Request官方文檔建議對(duì)所有的代碼設(shè)置超時(shí)。
如果你的python程序是同步的，忘記設(shè)置請(qǐng)求的默認(rèn)timeout可能會(huì)導(dǎo)致你的請(qǐng)求或者有應(yīng)用程序掛起。
timeout的設(shè)定同樣有兩種方法：
1、每次都在get語(yǔ)句中指定timeout的值。
（不可取，只對(duì)本次請(qǐng)求有效）。
requests.get('https://github.com/', timeout=0.001)
2、使用Transport Adapters設(shè)置統(tǒng)一的timeout時(shí)間（使用Transport Adapters，我們可以為所有HTTP調(diào)用設(shè)置默認(rèn)超時(shí)，這確保了即使開(kāi)發(fā)人員忘記在他的單個(gè)調(diào)用中添加timeout=1參數(shù)，也可以設(shè)置一個(gè)合理的超時(shí)，但這是允許在每個(gè)調(diào)用的基礎(chǔ)上重寫(xiě)。）：
下面是一個(gè)帶有默認(rèn)超時(shí)的自定義Transport Adapters的例子，在構(gòu)造http client和send()方法時(shí)，我們重寫(xiě)構(gòu)造函數(shù)以提供默認(rèn)timeout，以確保在沒(méi)有提供timeout參數(shù)時(shí)使用默認(rèn)超時(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)
實(shí)際代碼中，我們可以這樣使用：
import requests
http = requests.Session()
# 此掛載對(duì)http和https都有效
adapter = TimeoutHTTPAdapter(timeout=2.5)
http.mount("https://", adapter)
http.mount("http://", adapter)
# 設(shè)置默認(rèn)超時(shí)為2.5秒
response = http.get("https://api.twilio.com/")
# 通常為特定的請(qǐng)求重寫(xiě)超時(shí)時(shí)間
response = http.get("https://api.twilio.com/", timeout=10)
失敗時(shí)重試
網(wǎng)絡(luò)連接有丟包、擁擠，服務(wù)器出現(xiàn)故障。如果我們想要構(gòu)建一個(gè)真正健壯的程序，我們需要考慮失敗重試策略。
向HTTP client添加重試策略非常簡(jiǎn)單。創(chuàng)建一個(gè)HTTPAdapter來(lái)適應(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]允許重試的請(qǐng)求方法
method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]兩次重試的間隔參數(shù)
backoff_factor=0
合并timeouts和retries--超時(shí)與重試
綜合上面學(xué)到的，我們可以通過(guò)這種方法將timeouts與retries結(jié)合到同一個(gè)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請(qǐng)求
如果一個(gè)HTTP請(qǐng)求失敗了，可以用下面兩種方法獲取失敗的信息：
使用內(nèi)置的調(diào)試日志
使用request hooks
打印HTTP頭部信息
將logging debug level設(shè)置為大于0的值都會(huì)將HTTP請(qǐng)求的頭部打印在日志中。當(dāng)返回體過(guò)大或?yàn)樽止?jié)流不便于日志時(shí)，打印頭部將非常有用。
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)容
當(dāng)API返回內(nèi)容不太大時(shí)，我們可以使用request hooks與requests_toolbelt的dump工具庫(kù)輸出所有HTTP請(qǐng)求相應(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
測(cè)試與模擬請(qǐng)求
測(cè)試第三方API有時(shí)不能一直發(fā)送真實(shí)的請(qǐng)求（比如按次收費(fèi)的接口，還有沒(méi)開(kāi)發(fā)完的=_=），測(cè)試中我們可以用getsentry/responses作為樁模塊攔截程序發(fā)出的請(qǐng)求并返回預(yù)定的數(shù)據(jù)，造成返回成功的假象。
class TestAPI(unittest.TestCase):
@responses.activate #攔截該方法中的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è)定過(guò)的URL發(fā)請(qǐng)求了，不然會(huì)報(bào)錯(cuò)。
模仿瀏覽器行為
有些網(wǎng)頁(yè)會(huì)根據(jù)不同瀏覽器發(fā)送不同HTML代碼（為了反爬或適配設(shè)備），可以在發(fā)送請(qǐng)求時(shí)指定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"
})
綜上所有的點(diǎn)，實(shí)際使用是這樣子的：
self._enable_https = trueself.host = xxxxself.port = xxxxclass XxxxxXxxxx(object):def _get_api_session(self, timeout=30):address_prefix = 'http://'if self._enable_https:address_prefix = 'https://'#設(shè)置URLsess = sessions.BaseUrlSession(base_url=f"{address_prefix}{self.host}:{self.port}")#設(shè)置hooksassert_status_hook = lambda response, *args, **kwargs: response.raise_for_status()sess.hooks["response"] = [assert_status_hook]# 重試retries = Retry(total=3, backoff_factor=1, status_forcelist=[429])sess.mount(address_prefix, TimeoutHTTPAdapter(max_retries=retries, timeout=timeout))return sessclass TimeoutHTTPAdapter(HTTPAdapter):def __init__(self, *args, **kwargs):self.timeout = 5if "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.timeoutreturn super().send(request, **kwargs)
總結(jié)：
以上就是Python-Requests庫(kù)的進(jìn)階用法，在實(shí)際的代碼編寫(xiě)中將會(huì)很有用，不管是開(kāi)發(fā)編寫(xiě)API還是測(cè)試在編寫(xiě)自動(dòng)化測(cè)試代碼，都會(huì)極大的提高所編寫(xiě)代碼的穩(wěn)定性，服務(wù)穩(wěn)定了，升職加薪不是夢(mèng)，哈哈哈。