首頁 / 博客 / Python 與代理開發
【Python 實戰】使用 Requests 接入輪換住宅代理：會話、重試與逾時

【Python 實戰】使用 Requests 接入輪換住宅代理：會話、重試與逾時

分類：Python 與代理開發 釋出時間：2026-07-22 00:00 作者：SolisProxy 瀏覽：6

Python 透過住宅代理存取已授權的 HTTPS JSON API 時，可靠性取決於連線重用、逾時、有限重試與停止條件。以下教學核心已用本機 Python 3.13.3、Requests 2.33.1、urllib3 2.6.3 驗證，不含真實端點或憑證。

Session 與代理設定

requests.Session 使用 urllib3 連線池，重複存取同一主機時可重用連線。範例從 SOLIS_PROXY_URL 載入完整代理 URL，同時設定 http、https，並以 with 關閉 Session；代理 URL 不硬編碼或輸出。產品情境可參考住宅代理產品頁。

trust_env=False 會停用環境代理、NO_PROXY 與 .netrc，也忽略 REQUESTS_CA_BUNDLE、CURL_CA_BUNDLE。短例維持預設 TLS 驗證；正式環境如需私有 CA，應明確把 CA bundle 路徑設為 session.verify，禁止 verify=False。

Requests 預設不會逾時。timeout=(3.05, 20) 分別限制建連與相鄰資料讀取等待，但不是整體牆鐘時間。網路異常只輸出固定訊息，不列印原例外、URL、查詢參數或憑證。

只手動重試 GET 狀態

HTTPAdapter 的所有重試計數均為 0，transport 不會自動重送。應用層只有 session.get：初始請求加最多兩次重試，共最多三次，狀態限於 429、500、502、503、504。

Retry-After 由 urllib3 解析為秒數或 HTTP-date；無效值或等待超過 MAX_RETRY_WAIT=10 秒便停止，不提前重試。無標頭才採 0.5、1 秒有限指數退避。重試前先 close 回應；連線錯誤、讀取逾時、POST 與重新導向不重試。

HTTPS 目標與 JSON 契約

TARGET_URL 必須使用 HTTPS、有效連接埠為 443、不得含使用者資訊；正規化 hostname 須精確位於 TARGET_ALLOWED_HOSTS。urlsplit、hostname 或 port 發生 ValueError 時只產生固定 SystemExit，不回顯原值。程式不跟隨重新導向。

成功回應必須是 application/json 並解析成 dict 或 list；200 HTML 登入／CAPTCHA、無效 JSON 或純量值都停止。這不能辨識所有 CAPTCHA，正式業務仍須依 API 契約做嚴格結構、必要欄位、型別及錯誤信封驗證。

import json, os, sys, time
from urllib.parse import urlsplit
import requests
from requests.adapters import HTTPAdapter
from requests.exceptions import ConnectionError, ConnectTimeout, HTTPError, ProxyError, ReadTimeout, RequestException
from urllib3.exceptions import InvalidHeader
from urllib3.util import Retry
MAX_ATTEMPTS, MAX_RETRY_WAIT = 3, 10
RETRY_STATUSES = {429, 500, 502, 503, 504}
class RetryPolicyError(Exception): pass
class ResponsePolicyError(Exception): pass
def safe_split(value: str, name: str):
    try:
        parsed = urlsplit(value)
        host = (parsed.hostname or "").lower().rstrip(".")
        port = parsed.port
    except ValueError:
        raise SystemExit(f"{name} is invalid") from None
    return parsed, host, port
def require_target() -> str:
    value = os.environ.get("TARGET_URL", "").strip()
    parsed, host, port = safe_split(value, "TARGET_URL")
    allowed = {x.strip().lower().rstrip(".") for x in os.environ.get("TARGET_ALLOWED_HOSTS", "").split(",") if x.strip()}
    if parsed.scheme.lower() != "https" or not host or parsed.username or parsed.password:
        raise SystemExit("TARGET_URL is invalid") from None
    if (443 if port is None else port) != 443 or host not in allowed:
        raise SystemExit("TARGET_URL is not allowed") from None
    return value

def require_proxy() -> str:
    value = os.environ.get("SOLIS_PROXY_URL", "").strip()
    parsed, host, _ = safe_split(value, "SOLIS_PROXY_URL")
    if parsed.scheme.lower() not in {"http", "https"} or not host:
        raise SystemExit("SOLIS_PROXY_URL is invalid") from None
    return value
def build_session(proxy_url: str) -> requests.Session:
    disabled = Retry(total=0, connect=0, read=0, redirect=0, status=0, other=0)
    adapter = HTTPAdapter(max_retries=disabled)
    session = requests.Session()
    session.trust_env = False
    session.proxies.update({"http": proxy_url, "https": proxy_url})
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session
def retry_delay(response: requests.Response, attempt: int) -> float:
    raw = response.headers.get("Retry-After")
    if raw is None:
        return 0.5 * (2 ** attempt)
    try:
        delay = Retry(total=0).parse_retry_after(raw)
    except InvalidHeader:
        raise RetryPolicyError("invalid Retry-After") from None
    if delay > MAX_RETRY_WAIT:
        raise RetryPolicyError("Retry-After too long")
    return delay
def get_json(session: requests.Session, target_url: str):
    for attempt in range(MAX_ATTEMPTS):
        response = session.get(target_url, timeout=(3.05, 20), allow_redirects=False)
        try:
            if 300 <= response.status_code < 400:
                raise ResponsePolicyError("redirect rejected")
            if response.status_code in RETRY_STATUSES and attempt + 1 < MAX_ATTEMPTS:
                delay = retry_delay(response, attempt)
                response.close()
                time.sleep(delay)
                continue
            response.raise_for_status()
            media_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
            if media_type != "application/json":
                raise ResponsePolicyError("JSON required")
            try:
                payload = json.loads(response.content)
            except (UnicodeDecodeError, json.JSONDecodeError):
                raise ResponsePolicyError("invalid JSON") from None
            if not isinstance(payload, (dict, list)):
                raise ResponsePolicyError("unexpected JSON structure")
            return payload
        finally:
            response.close()
    raise AssertionError("unreachable")
def fail(message: str) -> None: print(message, file=sys.stderr)
def main() -> None:
    try:
        with build_session(require_proxy()) as session:
            payload = get_json(session, require_target())
            print(f"ok: {type(payload).__name__}")
    except ProxyError: fail("proxy connection failed; stopped")
    except ConnectTimeout: fail("connection timed out; stopped")
    except ReadTimeout: fail("response read timed out; stopped")
    except HTTPError: fail("HTTP status failed; stopped")
    except ConnectionError: fail("connection failed; stopped")
    except (RetryPolicyError, ResponsePolicyError): fail("response policy stopped")
    except RequestException: fail("request failed; stopped")
if __name__ == "__main__": main()

錯誤分類與會話選擇

範例明確捕捉 ProxyError、ConnectTimeout、ReadTimeout、HTTPError，再以 ConnectionError、RequestException 安全兜底。Requests 在收到標頭後讀正文逾時，有時會包裝成 ConnectionError；短例不宣稱精確區分，只會安全停止。

Session 的連線重用可能影響按連線輪換：若供應商以連線為輪換單位，多次請求可能沿用同一出口。實際輪換單位、sticky 格式與維持時間以已授權控制台為準，不猜測 SolisProxy 端點；可參考住宅代理會話控制說明。

生產環境檢查清單

環境變數只避免把憑證寫入原始碼；正式環境應使用密鑰管理器、最小權限與輪替機制。短例也不提供累計正文上限與 stream 讀取、軟／硬 deadline、自訂 CA、結構化日誌過濾，以及路徑、DNS rebinding、解析後 IP 與 egress 防火牆保證；部署時須按風險補齊，禁止 verify=False。

出口 IP 只用有權存取的回顯 API 或 SolisProxy 現有IP 位置查詢頁驗證。遵守目標條款、API／robots 政策、個資與所在地法律，限制併發及速率。連續錯誤、逾時、429、403、登入／CAPTCHA、預算或流量門檻都應停止並人工檢查；不得靠無限重試、加速或換身分繞過限制。

官方參考