PythonからHTTPを送る方法を、Requestsを中心にcurl・標準ライブラリ・HTTPXと比較します。待ち時間、応答status、JSON解析、再試行は別々の判断です。

合成データの教材はPython3.12.10 / Requests2.34.2 / urllib3 2.8.0 / HTTPX0.28.1で確認しました。実APIキー・外部API・DBは使わず、loopback HTTPだけを送っています。環境への導入は仮想環境と利用中の依存定義に合わせ、既存プロジェクトへ最新版を無条件に上書きしません。

この記事の要点

Python から HTTP を叩くなら requests。pip install requests で入る 標準ライブラリだけで済ませたいときは urllib.request。非同期や HTTP/2 が要るなら httpx JSON を送るのは json=、フォーム送信は data=。両者で Content-Type が変わる timeout を必ず指定する。既定はtimeoutなし。指定値は接続/読み取り待ちであり、総時間制限ではない Requestsでは4xx/5xxの確認にraise_for_status()を使う。JSON解析成功とは別

curl コマンドとの対応

curl	requests
curl URL	requests.get(url)
curl -X POST -d "a=1"	requests.post(url, data={"a": 1})
curl -H "Content-Type: application/json" でボディを送る	requests.post(url, json={...})
curl -H "Authorization: Bearer X"	headers={"Authorization": "Bearer X"}
curl -u user:pass	auth=("user", "pass")
curl -L	RequestsのGET等は既定で追従。head()は既定で追わない。切るならallow_redirects=False
curl --max-time 10	同義の指定ではありません。Requestsのtimeout=10は接続/読み取りの待ち時間で、全体10秒ではありません。
curl -F "file=@x.png"	files={"file": open("x.png", "rb")}
curl -k	verify=Falseは検証を無効化する対比で、推奨手順ではありません。開発でも既定検証か確認済みCAを使う

GET

旧コードの注意：次は未接続のAPI例です。example.comは教材用で、200やJSONのdictが返る保証はありません。Bearer TOKENを実値へ置換して公開せず、status・形式・サイズ・JSON解析を別々に確認します。timeout=10は全体を10秒で終了する指定ではありません。

import requests

res = requests.get(
    "https://api.example.com/users",
    params={"page": 2, "limit": 50},        # ?page=2&limit=50 に組み立てられる
    headers={"Authorization": "Bearer TOKEN"},
    timeout=10,                             # 必ず指定する
)

res.raise_for_status()                      # 4xx / 5xx なら例外
data = res.json()                           # JSON を dict に
print(res.status_code)                      # 200
print(res.headers["Content-Type"])
print(len(data))

params に辞書を渡せば URL エンコードは自動です。日本語や記号を自分で組み立てる必要はありません。

追加例：期待したJSONだけを上限付きで読む

fetch_json.pyは管理者が決めたURL向けの小さな同期教材です。sessionをwithで作り、value = fetch_json(session, configured_url)として呼びます。TLS検証を変更せず、redirectを追わず、200だけを受け付けます。APIが201や204を返す設計なら、その契約に合わせて別に実装します。URLの許可リストやDNS/接続先の検証を行うSSRF防御関数ではありません。

import json

import requests


class FetchError(Exception):
    pass


def fetch_json(session, url, *, limit=1048576, timeout=(3.05, 10)):
    if type(limit) is not int or not 1 <= limit <= 1048576:
        raise ValueError("limit must be 1..1048576")
    try:
        with session.get(url, timeout=timeout, stream=True,
                         allow_redirects=False) as response:
            if response.status_code != 200:
                raise FetchError("unexpected_status")
            media = response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower()
            if not (media == "application/json" or
                    (media.startswith("application/") and media.endswith("+json"))):
                raise FetchError("unexpected_media_type")
            body = bytearray()
            for chunk in response.iter_content(chunk_size=8192):
                if len(body) + len(chunk) > limit:
                    raise FetchError("body_limit")
                body.extend(chunk)
    except requests.exceptions.SSLError:
        raise FetchError("tls_error") from None
    except requests.exceptions.Timeout:
        raise FetchError("timeout") from None
    except requests.exceptions.RequestException:
        raise FetchError("transport_error") from None

    def reject_constant(value):
        raise ValueError("non_standard_json_constant")

    try:
        return json.loads(body.decode("utf-8"), parse_constant=reject_constant)
    except (UnicodeError, ValueError):
        raise FetchError("invalid_json") from None

JSON系のContent-Typeを確認し、iter_contentが返す展開後bytesを最大1MiBまで蓄積してUTF-8のJSONを解析します。null/配列/数値もJSONとして許可し、dictを前提にしません。非標準のNaN等は拒否します。上限は本文の蓄積量で、圧縮デコーダーやJSONオブジェクトを含むプロセスメモリの厳密な上限ではありません。timeoutも総時間を制限しません。

FetchErrorは固定の分類文字列だけを持ち、本文/URL/params/認証情報/例外文字列を含めません。呼出し側もdataや元例外をログへ出さない運用が必要で、既存のdebugログ/プロキシログ/監視をこの関数が匿名化するわけではありません。相手の業務結果や再送の安全性も決めません。

POST（JSON とフォーム）

旧コードの注意：次は別々の送信例で、urlやファイルは呼出し側の管理対象です。json/data/filesを同時指定すると意図した形式にならない場合があります。multipartのContent-Type/boundaryはライブラリに任せ、認証/課金/変更APIへ確認目的のPOSTを送りません。

# JSON で送る（Content-Type: application/json が自動で付く）
res = requests.post(
    "https://api.example.com/users",
    json={"name": "田中", "age": 30},
    timeout=10,
)

# フォーム送信（application/x-www-form-urlencoded）
res = requests.post(url, data={"name": "田中"}, timeout=10)

# 生のボディを送る
res = requests.post(url, data="raw text".encode("utf-8"),
                    headers={"Content-Type": "text/plain"}, timeout=10)

# ファイルアップロード（multipart/form-data）
with open("photo.png", "rb") as f:
    res = requests.post(url, files={"file": ("photo.png", f, "image/png")},
                        data={"title": "写真"}, timeout=10)

json=はJSON、data=の辞書はフォーム、data=のbytesは生の本文です。形式を取り違えた場合のstatusは相手API次第で、400とは限りません。送信Content-Typeと相手の仕様を照合し、filesによるmultipartのboundaryを手動で固定しません。教材では合成名のJSON/フォーム/生本文/ファイル送信をloopbackの受信側で比較しました。

レスポンスの中身

旧コードの注意：次のprint群は合成データだけで試す旧確認例です。text/contentは本文を全取得し、json()はdict以外も返します。url・本文・ヘッダーは秘密値を含み得ます。elapsedは応答ヘッダー受信までで、本文の全ダウンロード時間ではありません。

res = requests.get(url, timeout=10)

print(res.status_code)   # 200
print(res.ok)            # 400 未満なら True
print(res.text)          # 文字列（res.encoding に従ってデコード）
print(res.content)       # bytes（画像・PDF などはこちら）
print(res.json())        # JSON をパース。失敗すると JSONDecodeError
print(res.url)           # 最終的にアクセスした URL
print(res.elapsed)       # かかった時間

# 文字化けするときはエンコーディングを明示
res.encoding = "utf-8"
print(res.text)

詳しくは Responseオブジェクトの中身の確認 を参照してください。

エラー処理とタイムアウト

旧コードの注意：次の旧例は例外本文やHTTP本文をそのまま出すため、実APIのログ処理として使いません。先頭200文字だけでも匿名化にはなりません。SSLErrorはConnectionErrorの系統で、本文のstream読み取り中のtimeoutはConnectionErrorとして現れる場合もあります。固定分類だけを返す追加例と区別します。

import requests

try:
    res = requests.get(url, timeout=(3.0, 10.0))   # (接続, 読み取り) を別々に
    res.raise_for_status()
    data = res.json()
except requests.Timeout:
    print("時間内に応答がありませんでした")
except requests.ConnectionError:
    print("接続できませんでした（DNS / ネットワーク）")
except requests.HTTPError as e:
    print(f"HTTP エラー {e.response.status_code}: {e.response.text[:200]}")
except requests.JSONDecodeError:
    print("JSON として解釈できませんでした")
except requests.RequestException as e:
    print(f"その他の失敗: {e}")

Requestsの既定はtimeoutなしです。timeout=(接続, 読み取り)を明示しても、DNS・複数アドレスへの接続・リダイレクト・再試行/待機を含む総経過時間の厳密な上限ではありません。少量ずつ応答が来る相手にはread timeout以内でも全体が長くなり得ます。総時間の締切が必要なら、中断できる実行単位と接続/読取/再試行/待機を含めた別設計が必要です。

Session で接続を使い回す

旧コードの注意：次のSession例は既定ヘッダーとCookieを共有する断片です。任意のURLへ認証ヘッダーを使い回さず、利用者ごとの状態を混ぜません。Retryの既定対象メソッドにはGET以外も含まれます。再送可否・待機上限・結果不明を設計し、例外時もwithでSessionを閉じます。

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

session = requests.Session()
session.headers.update({"Authorization": "Bearer TOKEN"})   # 毎回付く

retry = Retry(total=3, backoff_factor=1,
              status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))

for page in range(1, 6):
    res = session.get(url, params={"page": page}, timeout=10)
    res.raise_for_status()

session.close()

Sessionは接続poolとヘッダー/Cookie等の状態を持ちます。本文を読み切るか応答を閉じることも接続再利用に関係し、毎回必ず高速になる保証ではありません。共有先/利用者を限定しwithで終了します。Cookieの範囲と保存はcookieの値の設定と取得を参照してください。

追加例：再試行を狭く限定する

retry_session.pyは読取専用と確認できた相手のGET/HEADと503だけを、初回＋最大2回まで試す教材です。接続/読取/その他エラーの自動再送は0、status最終失敗もResponseで返します。使うときはwith make_read_session() as session:とし、timeoutとallow_redirects=False、期待statusの確認も呼出し側で行います。

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


def make_read_session():
    policy = Retry(
        total=2, connect=0, read=0, redirect=0, other=0, status=2,
        allowed_methods=frozenset({"GET", "HEAD"}),
        status_forcelist=(503,), backoff_factor=0,
        respect_retry_after_header=False, raise_on_status=False,
    )
    session = requests.Session()
    for scheme in ("https://", "http://"):
        session.mount(scheme, HTTPAdapter(max_retries=policy))
    return session

教材では待機を0にしRetry-Afterも採用しません。運用APIのレート制限を無視してよいという意味ではなく、429/Retry-After/指数待機/jitter/総試行予算は契約に合わせて別途設計します。旧例の既定メソッド集合にはPUT/DELETE等もあり、「POST以外なら安全」とは言えません。GETにも相手実装による副作用があり、変更/課金操作は冪等キーと重複防止、成功したか不明な場合の照会手順が必要です。read失敗やtimeoutを未実行の証拠にしません。

標準ライブラリだけで済ませる

旧コードの注意：次は標準ライブラリの別例です。urlopenのtimeoutも総経過時間の上限ではありません。HTTPErrorはファイルに似た応答でもあるので利用後に閉じます。応答を無制限にreadする例を、大きな応答やユーザー入力URLにそのまま使いません。

import json
import urllib.parse
import urllib.request

# GET
url = "https://api.example.com/users?" + urllib.parse.urlencode({"page": 2})
req = urllib.request.Request(url, headers={"User-Agent": "my-app/1.0"})
with urllib.request.urlopen(req, timeout=10) as res:
    data = json.loads(res.read().decode("utf-8"))

# POST（JSON）
body = json.dumps({"name": "田中"}).encode("utf-8")
req = urllib.request.Request(url, data=body, method="POST",
                             headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=10) as res:
    print(res.status)

追加インストールができない環境ではこちらを使います。urlopen は 4xx / 5xx で urllib.error.HTTPError を投げる点が requests と違います。

httpx（非同期・HTTP/2）

旧コードの注意：次の旧非同期例は全URLのtaskを一度に作り、本文も通常取得します。件数/接続数/応答サイズを設計せずに大量URLへ広げません。HTTPXは既定でリダイレクトを追わず、HTTP/2も既定無効です。Requestsと例外型・timeout・送信APIが完全互換ではありません。

import asyncio
import httpx

async def main(urls):
    async with httpx.AsyncClient(timeout=10) as client:
        results = await asyncio.gather(*(client.get(u) for u in urls))
    for r in results:
        print(r.status_code, r.url)

asyncio.run(main(["https://example.com", "https://example.org"]))

HTTPXはRequestsの完全な置換ではありません。follow_redirects=Falseが既定で、raise_for_status()は3xxも例外の対象です。connect/read/write/poolのtimeoutを分け、HTTP/2は追加依存とhttp2=Trueおよび相手側対応が必要です。並列化だけで安全・高速を保証せず、相手の制限と処理の依存関係を確認します。

追加例：件数を限定しstatusだけを得る

async_statuses.pyは設定済みURLを最大8件、同時2件まで扱い、本文を読み込まずstatusの配列を返す教材です。redirectを追わずHTTP/2も無効と明示しています。asyncio.run(fetch_statuses(configured_urls))で呼び、既存のevent loop内ならawaitします。通信例外は呼出し側へ伝わり、gatherを使った全件の業務結果保証やSSRF防御ではありません。

import asyncio
from itertools import islice

import httpx


async def fetch_statuses(urls):
    urls = tuple(islice(urls, 9))
    if len(urls) > 8:
        raise ValueError("at most eight configured URLs")
    gate = asyncio.Semaphore(2)
    timeout = httpx.Timeout(10, connect=3)
    limits = httpx.Limits(max_connections=2, max_keepalive_connections=2)
    async with httpx.AsyncClient(timeout=timeout, limits=limits,
                                 follow_redirects=False, http2=False) as client:
        async def one(url):
            async with gate:
                async with client.stream("GET", url) as response:
                    return response.status_code

        return await asyncio.gather(*(one(url) for url in urls))

安全面の注意

ユーザー入力URLはscheme/host/portだけでなく、名前解決後の宛先、redirect先、proxy経路や内部ネットワークへの到達も含めて制限します。単純な文字列prefix検査だけでSSRF対策済みとしません。Requests/HTTPXは環境のproxyや証明書設定等も採用するため、診断のために本番の組織proxyを無断で迂回しません。

• 証明書検証を維持します。社内CA等は管理者が確認したCA bundleをverifyへ指定し、名前/SAN・期限・チェーン・プロキシ経路を照合します。開発環境でもverify=Falseを通常の解決策にしません。証明書名不一致の切り分けも参照してください。
• トークンをソースに直書きせず、環境変数（os.environ["API_TOKEN"]）から読む
• ユーザー入力をそのまま URL にしない。アクセス先を限定する
• stream=Trueだけではサイズ上限になりません。iter_contentで読む量を数え、超過で停止して応答を閉じます。Content-Lengthだけを信頼せず、圧縮後/展開後の量、デコーダー/JSON解析のメモリも別に考えます。

関連

• 文法 — 親カテゴリ
• Responseオブジェクトの中身の確認
• cookieの値の設定と取得
• 外部ライブラリ — requests の導入
• 例外処理
• Webスクレイピング

確認した範囲と公式資料（2026年10月26日）

実HTTPでJSON/フォーム/生本文/multipart、redirect、status/形式/不正JSON/空/展開後サイズ、read timeout、少量連続応答が総時間を超えて続くこと、elapsedと本文取得の差、GET503の最大3試行/POST1試行、標準urllibのHTTPError、HTTPXの3xxと限定async例を検査しました。TLS失敗の分類だけは模擬例外で、実証明書/HTTPS/proxy・実認証・HTTP/2・本番APIの副作用/待機/負荷やSSRF防御全体は未検証です。

公式資料：Requestsの送信/JSON/timeout、Session・TLS・stream、RequestsのResponse/例外、urllib3 Retry、urllib.request、HTTPX互換性、HTTPX timeout、HTTP/2の有効化、curlのmax-time。