Background
使用 Python 呼叫某第三方服務的 report API，希望取得一些資料來做廣告成效報表，初步使用 urllib3 來實作發送 HTTP 請求
遇到的問題
- 在 Local (Laptop) 環境，請求 API 失敗（返回 400 Error）
- 在 Production (Jenkins) 環境，第一次成功，之後請求失敗（返回 400 Error）
Issue Analysis
使用 urllib3 時常失敗，但使用 curl 測試請求，發現可以穩定成功。
- 代表 API 可能對
urllib3和curl回應的行為是不同的 - 懷疑 API 服務對 Header 有流量或其他限制，另外 Return 400 error 感覺是工程師亂寫的，參考就好
嘗試方法：
- 使用
urllib3, 改User-agent可能有用 - 可能的原因：實際上也有很多 API 服務，為了防爬蟲，會針對常見的 library 的 user-agent or connection 的方式有阻擋限制
- 直接改用
pycurl方式連線 - 可能的原因：urllib3 和 curl 的底層連線方式不同
pycurl vs. urllib3
Solution
- 先在 urllib3 裡指定 User-Agent 到 curl, 但仍然不 work.
- 將原本
urllib3的實作，改為pycurl的實作方式
- urllib3
# Default Header
{
"headers": {
"Host": "httpbin.org",
"Accept-Encoding": "identity",
"User-Agent": "python-urllib3/1.26.16",
"Accept": "*/*",
"Connection": "close"
}
}# Example
import urllib3
from urllib3.util.retry import Retry
from urllib3.exceptions import HTTPError
import time
# 設定重試機制
retry_strategy = Retry(
total=5, # 總共重試 5 次
backoff_factor=1, # 每次失敗後等待時間 (1s, 2s, 4s, 8s, 16s...)
status_forcelist=[500, 502, 503, 504], # 這些錯誤碼時才會重試
allowed_methods=["GET"], # 只對 GET 方法啟用重試
)
# 建立連接池並套用重試策略
http = urllib3.PoolManager(retries=retry_strategy)
# 發送請求並加上錯誤處理
def fetch_data(url):
try:
response = http.request("GET", url)
response.raise_for_status() # 如果 HTTP 狀態碼錯誤，會拋出例外
print(f"Status Code: {response.status}")
print("Response Body:", response.data.decode("utf-8"))
except HTTPError as e:
print(f"HTTP 錯誤: {e}")
except Exception as e:
print(f"其他錯誤: {e}")
# 執行 API 請求
fetch_data(url = "<https://jsonplaceholder.typicode.com/posts/1>")
- pycurl
# Default header
{
"headers": {
"Host": "httpbin.org",
"User-Agent": "PycURL/7.43.0.6 libcurl/7.68.0 OpenSSL/1.1.1f zlib/1.2.11",
"Accept": "*/*"
}
}
# Example
import pycurl
import certifi
import io
import time
# 最大重試次數
MAX_RETRIES = 5
BACKOFF_FACTOR = 1 # 每次失敗後等待 (1s, 2s, 4s, 8s, 16s)
def fetch_data(url):
retries = 0
while retries < MAX_RETRIES:
try:
buffer = io.BytesIO() # 用來存放回應的資料
# 設定 pycurl 參數
curl = pycurl.Curl()
curl.setopt(pycurl.URL, url) # 設定 URL
curl.setopt(pycurl.WRITEFUNCTION, buffer.write) # 把回應寫入 buffer
curl.setopt(pycurl.CAINFO, certifi.where()) # 設定 SSL 憑證
curl.setopt(pycurl.FOLLOWLOCATION, True) # 自動跟隨重定向
# 執行請求
curl.perform()
# 獲取 HTTP 狀態碼
status_code = curl.getinfo(pycurl.RESPONSE_CODE)
curl.close()
if status_code == 200:
print(f"Status Code: {status_code}")
print("Response Body:", buffer.getvalue().decode("utf-8"))
return
else:
raise Exception(f"HTTP 錯誤: {status_code}")
except Exception as e:
retries += 1
wait_time = BACKOFF_FACTOR * (2 ** (retries - 1))
print(f"第 {retries} 次重試: {e}, 等待 {wait_time} 秒")
time.sleep(wait_time)
print("請求失敗，已達最大重試次數")
# 執行 API 請求
fetch_data(url = "<https://jsonplaceholder.typicode.com/posts/1>")
