[Python API debugging] 用 urllib3 呼叫 API, 有時失敗、有時成功

Alice Hsu

更新於 2025/02/16發佈於 2025/02/16閱讀時間約 9 分鐘

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>")