【保存版】Python requestsのtimeout設定パターンと実践レシピ

import requests

# timeoutを指定していないNG例
resp = requests.get("https://api.example.com/data")
print(resp.status_code)

import requests

# 両方に5秒のtimeoutを設定する例
response = requests.get(
    "https://httpbin.org/delay/3",  # 3秒後に応答を返すテストAPI
    timeout=5                       # connect + read に5秒を設定
)
print(response.status_code, response.elapsed.total_seconds())

import requests

# (connect_timeout, read_timeout) のタプルで指定
url = "https://httpbin.org/delay/3"  # 3秒遅延

response = requests.get(
    url,
    timeout=(2.0, 5.0)  # 接続は2秒まで、レスポンス読み取りは5秒まで
)

print("status:", response.status_code)
print("elapsed:", response.elapsed.total_seconds())

import requests

session = requests.Session()
session.headers.update({"User-Agent": "my-app/1.0"})

def fetch_with_default_timeout(url, timeout=(3.0, 5.0)):
    # 共通のtimeoutを関数でラップして使いやすくする
    return session.get(url, timeout=timeout)

resp = fetch_with_default_timeout("https://httpbin.org/get")
print(resp.status_code)

import requests
from requests.exceptions import Timeout, RequestException

API_TIMEOUT = (2.0, 6.0)  # (connect, read)

def call_external_api():
    url = "https://api.example.com/v1/resource"
    try:
        resp = requests.get(url, timeout=API_TIMEOUT)
        resp.raise_for_status()
        return resp.json()
    except Timeout:
        # 外部APIの遅延・障害
        # ログ出力やフォールバック処理などを行う
        return {"status": "timeout", "data": None}
    except RequestException as e:
        # その他の通信エラー
        return {"status": "error", "detail": str(e)}

result = call_external_api()
print(result)

import requests
from requests.exceptions import Timeout

BATCH_TIMEOUT = (5.0, 60.0)  # バッチ専用の長めtimeout

def download_large_report(job_id: str):
    url = f"https://report.example.com/jobs/{job_id}/result"
    try:
        resp = requests.get(url, timeout=BATCH_TIMEOUT)
        resp.raise_for_status()
        return resp.content  # バイナリ(例: PDF)
    except Timeout:
        # バッチジョブなら、リトライやステータス更新などを行う
        raise RuntimeError(f"Job {job_id} timed out while downloading report")

content = download_large_report("job-123")
print(len(content))

from flask import Flask, jsonify
import requests
from requests.exceptions import Timeout

app = Flask(__name__)

API_TIMEOUT = (1.0, 2.0)  # Webアプリ用に短めのtimeout

@app.route("/profile/<user_id>")
def profile(user_id):
    url = f"https://api.example.com/users/{user_id}"
    try:
        resp = requests.get(url, timeout=API_TIMEOUT)
        resp.raise_for_status()
        data = resp.json()
        return jsonify({"ok": True, "profile": data})
    except Timeout:
        # 一定時間待っても応答がない場合は、簡易情報のみ返すなどのフォールバックも検討
        return jsonify({
            "ok": False,
            "error": "timeout",
            "message": "現在プロフィール情報を取得できません。しばらくしてから再度お試しください。"
        }), 504

if __name__ == "__main__":
    app.run(debug=True)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry(
    total_retries=3,
    backoff_factor=0.5,
    status_forcelist=(500, 502, 503, 504),
    timeout=(1.0, 2.0),
):
    session = requests.Session()

    retry = Retry(
        total=total_retries,
        read=total_retries,
        connect=total_retries,
        backoff_factor=backoff_factor,  # 再試行間隔(指数的に増加)
        status_forcelist=status_forcelist,
        allowed_methods=["GET", "POST"],
        raise_on_status=False,
    )

    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)

    # timeoutはアダプタには設定できないので、呼び出し側で指定する
    session.request_timeout = timeout  # 自前プロパティとして保持しておく

    return session

session = create_session_with_retry()

try:
    resp = session.get(
        "https://httpbin.org/status/500",
        timeout=session.request_timeout
    )
    print(resp.status_code)
except requests.exceptions.RequestException as e:
    print("error:", e)

import requests
from requests.exceptions import Timeout, RequestException

def fetch(url, timeout=(2.0, 5.0)):
    try:
        resp = requests.get(url, timeout=timeout)
        resp.raise_for_status()
    except Timeout as e:
        # timeoutは明確に区別して扱う
        raise RuntimeError(f"Request to {url} timed out") from e
    except RequestException as e:
        # DNSエラー、接続拒否、HTTPエラーなど
        raise RuntimeError(f"Request to {url} failed: {e}") from e
    else:
        return resp.text

print(fetch("https://httpbin.org/get")[:80])

import logging
import time
import requests
from requests.exceptions import Timeout, RequestException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def fetch_with_retry(url, timeout=(1.0, 2.0), retries=2):
    for attempt in range(1, retries + 2):  # 初回 + リトライ回数
        start = time.time()
        try:
            resp = requests.get(url, timeout=timeout)
            resp.raise_for_status()
            elapsed = time.time() - start
            logger.info(
                "success url=%s status=%s elapsed=%.2f attempt=%s",
                url, resp.status_code, elapsed, attempt
            )
            return resp
        except Timeout as e:
            elapsed = time.time() - start
            logger.warning(
                "timeout url=%s timeout=%s elapsed=%.2f attempt=%s error=%s",
                url, timeout, elapsed, attempt, e
            )
        except RequestException as e:
            elapsed = time.time() - start
            logger.error(
                "request_error url=%s elapsed=%.2f attempt=%s error=%s",
                url, elapsed, attempt, e
            )
            break  # リトライしても無駄そうな場合は中断
    raise RuntimeError(f"Failed to fetch {url} after {retries + 1} attempts")

fetch_with_retry("https://httpbin.org/delay/3", timeout=(1.0, 1.0), retries=1)

from flask import Flask, jsonify
import logging
import requests
from requests.exceptions import Timeout

app = Flask(__name__)
logger = logging.getLogger(__name__)

@app.route("/search")
def search():
    query = "example"
    url = f"https://api.example.com/search?q={query}"
    timeout = (1.0, 2.0)

    try:
        resp = requests.get(url, timeout=timeout)
        resp.raise_for_status()
        return jsonify(resp.json())
    except Timeout as e:
        # 内部ログ(詳細)
        logger.warning("search timeout url=%s timeout=%s error=%s", url, timeout, e)

        # ユーザー向けレスポンス(簡潔)
        return jsonify({
            "ok": False,
            "error": "timeout",
            "message": "検索結果の取得に時間がかかっています。時間をおいて、もう一度お試しください。"
        }), 504

if __name__ == "__main__":
    app.run()

# http_client.py
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

DEFAULT_TIMEOUT = (2.0, 5.0)

class TimeoutHTTPAdapter(HTTPAdapter):
    """デフォルトtimeoutを持つHTTPAdapter"""

    def __init__(self, *args, timeout=DEFAULT_TIMEOUT, **kwargs):
        self.timeout = 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)

def create_http_session(timeout=DEFAULT_TIMEOUT, retries=2):
    session = requests.Session()
    retry = Retry(
        total=retries,
        backoff_factor=0.3,
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
    )
    adapter = TimeoutHTTPAdapter(max_retries=retry, timeout=timeout)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

# 使用側のコード
from http_client import create_http_session

session = create_http_session()

# timeoutを明示しなくてもDEFAULT_TIMEOUTが適用される
resp = session.get("https://httpbin.org/get")
print(resp.status_code)

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_optimized_session():
    session = requests.Session()

    retry = Retry(
        total=3,
        connect=3,
        read=3,
        backoff_factor=0.2,
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
    )

    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=20,  # 接続プール数
        pool_maxsize=50,      # 各プールの最大接続数
    )

    session.mount("http://", adapter)
    session.mount("https://", adapter)

    return session

session = create_optimized_session()
response = session.get("https://httpbin.org/get", timeout=(2.0, 5.0))
print(response.status_code)