快轉到主要內容
  1. 教學文章/

Python Requests 實戰:Session、Timeout、Retry 與可靠 API Client

·7 分鐘· loading · loading · ·
Python Requests HTTP API Session Retry Developer-Tools
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 118: 本文

featured

一. 前言:會送出 Request,不等於會寫可靠的 Client
#

Python 要呼叫 HTTP API,很多人的第一反應都是:

import requests
response = requests.get("https://api.example.com/items")
print(response.json())

這段很好懂,也正是 Requests 長年受歡迎的原因。 但正式程式只要多活幾天,問題就會慢慢冒出來:

  • 每次呼叫都重新建立連線
  • 忘了設 timeout,工作就一直卡著
  • 500 被當成正常 response 繼續處理
  • POST 失敗後無腦重送,結果建立兩筆資料
  • Token、base URL 與錯誤處理散落在各個檔案
  • 測試必須真的連外,CI 一斷網就一起倒下 所以這篇不只教 requests.get()。 拍拍君要帶你把 Requests 整理成一個可預期、可測試、會善後的同步 API client。 如果你的應用程式需要 async/await、HTTP/2 或高併發 I/O,請直接看 HTTPX 教學;今天專注在 Requests 最擅長的同步工作流。

二. 安裝:確認版本與執行環境
#

uv 建立專案:

mkdir pypy-requests-demo
cd pypy-requests-demo
uv init
uv add requests

或用既有虛擬環境安裝:

python -m pip install requests

確認版本:

import requests
print(requests.__version__)

本文查核時,PyPI 最新穩定版是 Requests 2.34.2,官方支援 Python 3.10 以上。 2.34 系列也把型別資訊直接放進套件,使用 mypy、Pyright 或 IDE 時更順手。 版本會繼續更新,部署時仍應由 lockfile 固定實際測過的版本:

uv lock
uv sync --locked

套件管理流程可以搭配 uv 完整教學 一起看。

三. 先把 Request 與 Response 邊界看清楚
#

最常見的 GET,建議把 query parameter 交給 params

import requests
response = requests.get(
    "https://httpbin.org/get",
    params={"q": "拍拍君", "page": 2},
    timeout=(3.05, 10),
)
print(response.url)
print(response.status_code)
print(response.headers.get("content-type"))
print(response.json()["args"])

Requests 會負責 URL encoding,不要自己用字串拼 ?q=...。 送 JSON 時使用 json=,不要把 dict 手動 json.dumps() 後塞進 data=

payload = {
    "name": "拍拍君",
    "enabled": True,
}
response = requests.post(
    "https://httpbin.org/post",
    json=payload,
    timeout=(3.05, 10),
)
response.raise_for_status()
data = response.json()
print(data["json"])

這樣 Requests 會替你序列化 JSON,並設定合適的 Content-Type

不要把 response.json() 當成成功證明
#

伺服器即使回 404500,body 仍可能是合法 JSON。 所以順序通常是:

response = requests.get(url, timeout=(3.05, 10))
response.raise_for_status()
data = response.json()

raise_for_status() 負責 HTTP status;json() 只負責解碼 body。 這兩個問題不要揉成一團。

四. Session:重用連線,也集中共用狀態
#

頂層的 requests.get() 適合一次性腳本。 同一個程式會連續呼叫同一個 host 時,請建立 Session

import requests
with requests.Session() as session:
    session.headers.update({
        "Accept": "application/json",
        "User-Agent": "pypy-client/1.0",
    })
    first = session.get(
        "https://httpbin.org/cookies/set/theme/green",
        timeout=(3.05, 10),
    )
    first.raise_for_status()
    second = session.get(
        "https://httpbin.org/cookies",
        timeout=(3.05, 10),
    )
    second.raise_for_status()
    print(second.json()["cookies"])

Session 做了兩件很實用的事:

  1. 保存 header、cookie、auth 等共用設定
  2. 透過 urllib3 的 connection pool 重用底層 TCP 連線 第二次請求不必每次重新做完整連線握手,密集呼叫同一服務時差異很明顯。

Session 是有狀態物件
#

方便也代表要管理生命週期。

  • with 或明確呼叫 close()
  • 不要把不同使用者的 cookie 混在同一個 Session
  • 多執行緒程式不要隨意共享並修改 Session 狀態
  • 密鑰放環境變數,不要硬寫進原始碼 一個服務 client 擁有一個 Session,通常比全專案共用單一全域 Session 更清楚。

五. Timeout:一定要設,而且它不是總秒數
#

Requests 預設沒有 timeout。 也就是說,這段可能等非常久:

# 不建議放進 production
requests.get("https://api.example.com/report")

單一數字會同時套用在 connect 與 read timeout:

requests.get(url, timeout=10)

更清楚的做法是二元素 tuple:

CONNECT_TIMEOUT = 3.05
READ_TIMEOUT = 20.0
response = requests.get(
    url,
    timeout=(CONNECT_TIMEOUT, READ_TIMEOUT),
)
  • connect timeout:建立連線最多等多久
  • read timeout:連線後,等待伺服器送出資料的間隔上限 請注意,read timeout 不是整份下載的 wall-clock deadline。 只要資料持續分段抵達,總下載時間可能超過設定值。

不要期待 Session 自動有預設 timeout
#

session.headers 可以設預設 header,Session 卻沒有簡單的全域 timeout 欄位。 拍拍君偏好在自己的 client 邊界補上:

class APIClient:
    def __init__(self, timeout: tuple[float, float] = (3.05, 20.0)):
        self.timeout = timeout
        self.session = requests.Session()
    def request(self, method: str, url: str, **kwargs):
        kwargs.setdefault("timeout", self.timeout)
        return self.session.request(method, url, **kwargs)

這樣每個呼叫都有預設值,特殊 endpoint 仍可覆寫。

六. 錯誤處理:分辨網路、HTTP 與資料錯誤
#

Requests 公開的例外都繼承自 RequestException。 實務上可以保留幾個重要邊界:

import requests
try:
    response = requests.get(url, timeout=(3.05, 10))
    response.raise_for_status()
    data = response.json()
except requests.Timeout as exc:
    raise RuntimeError("上游 API 回應逾時") from exc
except requests.ConnectionError as exc:
    raise RuntimeError("無法連上上游 API") from exc
except requests.HTTPError as exc:
    status = exc.response.status_code if exc.response is not None else None
    raise RuntimeError(f"上游回傳 HTTP {status}") from exc
except requests.exceptions.JSONDecodeError as exc:
    raise RuntimeError("上游回應不是合法 JSON") from exc

不要直接寫 except Exception: return {}。 那會讓 DNS 錯誤、權限錯誤、伺服器錯誤與資料格式錯誤全部變成同一個空 dict, 真正的 bug 反而更難找。 Client 可以把低階例外轉成專案自己的 domain error,但要用 raise ... from exc 保留原始 traceback。

七. Retry:用 HTTPAdapter 表達傳輸策略
#

Requests 預設不會替所有失敗自動重試。 要在 Session 層設定策略,可以把 urllib3 的 Retry 掛到 HTTPAdapter

import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
def build_session() -> requests.Session:
    retry = Retry(
        total=4,
        connect=4,
        read=2,
        status=3,
        allowed_methods=frozenset({"GET", "HEAD", "OPTIONS"}),
        status_forcelist=(429, 500, 502, 503, 504),
        backoff_factor=0.5,
        backoff_jitter=0.2,
        respect_retry_after_header=True,
        raise_on_status=False,
    )
    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=10,
        pool_maxsize=20,
        pool_block=True,
    )
    session = requests.Session()
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

重點不是數字抄幾次,而是政策要寫得出理由:

  • 只重試暫時性錯誤
  • 次數有限
  • 使用退避,避免服務恢復時一起湧入
  • 尊重 Retry-After
  • 預設只重試適合安全重送的方法

為什麼範例沒有放 POST?
#

因為 POST 常代表建立訂單、扣款、寄信或啟動工作。 第一次其實成功、只是回應在途中遺失時,重送可能造成重複副作用。 除非 API 有明確 idempotency key 契約,否則不要只為了「比較耐打」就重試 POST。 如果你要深入理解停止條件、指數退避、jitter、非 HTTP 工作與 async retry, 請看 tenacity 容錯教學。這裡只處理 Requests transport adapter 的責任。

八. PreparedRequest:檢查真正要送出去的內容
#

Requests 會先建立 Request,再轉成可以傳輸的 PreparedRequest。 大多數程式不必手動碰它;需要簽章或檢查最終 body 時才使用。 重點是透過 Session 準備,保留 cookie 與共用 header:

from requests import Request, Session
session = Session()
session.headers["User-Agent"] = "pypy-client/1.0"
request = Request("POST", "https://httpbin.org/post", json={"message": "拍拍君出發"})
prepared = session.prepare_request(request)
print(prepared.method, prepared.url, prepared.body)
settings = session.merge_environment_settings(prepared.url, {}, None, None, None)
response = session.send(prepared, timeout=(3.05, 10), **settings)
response.raise_for_status()

請用 session.prepare_request(),而不是裸 request.prepare()。 手動 send() 時也別漏掉 merge_environment_settings(),否則 CA bundle 或 proxy 等環境值可能不會套用。

九. TLS、Proxy 與密鑰:方便不能拿安全交換
#

Requests 預設會驗證 TLS 憑證,請保持這個預設。

response = session.get(
    "https://api.example.com/health",
    timeout=(3.05, 10),
    verify=True,
)

不要因為遇到憑證錯誤,就把 verify=False 永久留在 production。 正確處理方式通常是修正憑證鏈,或指定可信任 CA bundle:

response = session.get(
    url,
    verify="/path/to/company-ca.pem",
    timeout=(3.05, 10),
)

Proxy 可由參數或 HTTP_PROXYHTTPS_PROXYNO_PROXY 等環境設定提供。 Proxy URL 可能包含帳密,不能印進 log,也不能提交到 Git。 API token 同樣從環境讀取:

import os
token = os.environ["PYPY_API_TOKEN"]
session.headers["Authorization"] = f"Bearer {token}"

需要整理 .env 開發設定時,可參考 python-dotenv 教學, 但正式密鑰仍應交給部署平台的 secret 管理功能。

十. 組成一個小而完整的 API Client
#

現在把 base URL、Session、timeout、retry 與錯誤翻譯集中起來:

from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import requests
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
class PypyAPIError(RuntimeError):
    """拍拍 API 呼叫失敗。"""
@dataclass(frozen=True)
class ClientConfig:
    base_url: str
    token: str
    timeout: tuple[float, float] = (3.05, 15.0)
class PypyClient:
    def __init__(self, config: ClientConfig) -> None:
        self.config = config
        self.session = self._build_session()
    def _build_session(self) -> requests.Session:
        retry = Retry(
            total=3,
            allowed_methods=frozenset({"GET", "HEAD"}),
            status_forcelist=(429, 500, 502, 503, 504),
            backoff_factor=0.5,
            respect_retry_after_header=True,
            raise_on_status=False,
        )
        adapter = HTTPAdapter(max_retries=retry)
        session = requests.Session()
        session.headers.update({
            "Accept": "application/json",
            "Authorization": f"Bearer {self.config.token}",
            "User-Agent": "pypy-client/1.0",
        })
        session.mount("https://", adapter)
        return session
    def close(self) -> None:
        self.session.close()
    def __enter__(self) -> PypyClient:
        return self
    def __exit__(self, *args: object) -> None:
        self.close()
    def _request(self, method: str, path: str, **kwargs: Any) -> Any:
        url = f"{self.config.base_url.rstrip('/')}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.config.timeout)
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.Timeout as exc:
            raise PypyAPIError("API 回應逾時") from exc
        except requests.HTTPError as exc:
            status = (
                exc.response.status_code
                if exc.response is not None
                else "unknown"
            )
            raise PypyAPIError(f"API 回傳 HTTP {status}") from exc
        except requests.RequestException as exc:
            raise PypyAPIError("API 連線失敗") from exc
    def list_notes(self, limit: int = 20) -> list[dict[str, Any]]:
        data = self._request("GET", "/notes", params={"limit": limit})
        if not isinstance(data, list):
            raise PypyAPIError("API 回應格式錯誤:預期 list")
        return data

使用端只需要知道服務語意:

import os
config = ClientConfig(
    base_url="https://api.example.com/v1",
    token=os.environ["PYPY_API_TOKEN"],
)
with PypyClient(config) as client:
    for note in client.list_notes(limit=5):
        print(note["title"])

這個 client 故意保持同步、小型、責任單純。 如果需求成長到 OAuth refresh、分頁 iterator、schema validation 或觀測性, 再逐步加進明確的層,不必第一天就做成 SDK 宇宙戰艦。

十一. Requests、HTTPX、Tenacity 怎麼選?
#

需求 建議工具
簡單同步腳本、成熟 HTTP/1.1 生態 Requests
需要同步 Session、cookie、PreparedRequest Requests
原生 async、HTTP/2、細分 pool/write timeout HTTPX
非 HTTP 任務也需要複雜重試政策 Tenacity
Requests transport 層的有限重試 HTTPAdapter + urllib3 Retry
選擇工具不是比誰比較新。
同步 CLI、批次腳本與傳統 Web worker 使用 Requests 完全合理;
async service 硬塞 blocking Requests,才是真的在跟 event loop 作對。

結語
#

Requests 的 API 很簡單,但簡單不代表只能寫玩具程式。 只要把幾個邊界守好:

  • Session 管理狀態與連線池
  • timeout 限制每次等待
  • raise_for_status() 分開 HTTP 錯誤
  • HTTPAdapter 表達有限、保守的 retry
  • PreparedRequest 留給需要精確控制的場景
  • 自己的 client 集中設定、例外與測試介面 你就能保留 Requests 的好讀,同時避開「偶爾卡死、偶爾重複送、出了錯不知道在哪裡」的經典災情。 HTTP client 不需要寫得很炫。穩定、可觀測、能預期,才是真正的帥。

延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 118: 本文

相關文章

FastAPI + Streamlit 實戰:API 後端與互動前端分工
·9 分鐘· loading · loading
Python FastAPI Streamlit API Frontend Developer-Tools
Streamlit Data Editor 實戰:可編輯表格、上傳驗證與 CSV 匯入匯出
·8 分鐘· loading · loading
Python Streamlit Data-Editor CSV Validation Developer-Tools
uv 管理 Python 版本:Install、Find、Pin、Upgrade 與直譯器選擇
·9 分鐘· loading · loading
Python Uv Python Versions Interpreter Virtualenv Developer-Tools
Textual Form Wizard 實戰:多步驟表單、Validation 與狀態切換
·5 分鐘· loading · loading
Python Textual TUI Forms Validation Developer-Tools
uv + GitHub Actions 實戰:Lockfile、Cache 與可重現 CI
·5 分鐘· loading · loading
Python Uv GitHub Actions Ci Lockfile Cache Developer-Tools
Python OpenTelemetry 實戰:Trace、Span 與 FastAPI 觀測流程
·6 分鐘· loading · loading
Python OpenTelemetry Observability FastAPI Tracing Developer-Tools