PythonでのHTTPXプロキシ：セットアップと非同期処理ガイド | Geonode

httpxは、requestsでは対応しきれなくなった新しいPythonプロジェクトの多くが採用するHTTPクライアントです。その主な理由は、同期コードと非同期コードの両方に対応した単一のAPIが提供されているためです。このガイドでは、httpxプロキシをClientおよびAsyncClientの両方に組み込み、ローテーションされる一般家庭向けエンドポイントに対して認証を行い、スティッキーセッションを固定し、SOCKS5を追加し、タイムアウトと再試行を調整し、実際に発生する3つのエラーを診断する方法について説明します。このガイドを読み終える頃には、プロキシを経由してルーティングを行い、出口IPを証明できる再利用可能なクライアントが完成しているはずです。

サンプルコードは httpx 0.27 以降を対象としています。古いリリースを使用している場合は、httpx 0.26 で proxies= キーワードが proxy= および mounts= に名称変更された点に注意してください。以下の説明ではすべて現在の名称を使用しています。

httpxプロキシの設定方法：`proxy`と`mounts`

httpxには、シンプルな設定方法と詳細な設定方法の2種類があります。proxy=は単一のURLを指定し、すべてのリクエストをそのURL経由でルーティングします。 mounts=は、URLパターンをトランスポートにマッピングし、各トランスポートは独自のプロキシ、リトライ回数、接続制限を設定できます。

オプション	スコープ	使用場面
`proxy="http://..."`	クライアント全体、すべてのスキーム	すべてに1つのプロキシを使用します。これは、ローテーションゲートウェイで一般的なケースです
`mounts={"https://": HTTPTransport(proxy=...)}`	URLパターンごと	スキームやホストごとに異なるプロキシを使用する場合、または内部サービスに対してプロキシをバイパスする場合
`httpx.get(url, proxy=...)`	単一のトップレベル呼び出し	長期にわたり存続するクライアントを必要としない簡易なチェックやスクリプト

単純な形式は次のようになります：

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

with httpx.Client(proxy=PROXY, timeout=20.0) as client:
    r = client.get("https://api.ipify.org?format=json")
    print(r.status_code, r.json())

ポート 9000 は HTTP(S) ローテーションゲートウェイであるため、新しい接続ごとに異なる住宅用 IP が割り当てられます。認証情報は URL に Basic 認証として含まれており、httpx は追加のコードを一切書かずにそれらを Proxy-Authorization ヘッダーに変換します。

スキームごとのルーティングにおける注意点

mounts を使用すると、http:// と https:// のターゲットに別々のトランスポートを割り当てることができます。注意すべき点は、プロキシ URL 自体のスキームです。 https://というターゲットの場合、httpxはプロキシに対してCONNECTを発行し、そのトンネル内でオリジンとのTLSネゴシエーションを行います。プロキシへのホップはプレーンなHTTPであるため、https://マウント上のプロキシURLは通常、https://ではなくhttp://で始まる必要があります。そこに https:// と記述すると、httpx はプロキシへのホップ自体を TLS でラップするよう指示されますが、ほとんどのゲートウェイはこれに対応していないため、プロキシとは無関係に見えるハンドシェイクエラーが発生します。

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

mounts = {
    "http://": httpx.HTTPTransport(proxy=PROXY),
    "https://": httpx.HTTPTransport(proxy=PROXY),  # http:// scheme is correct here
    # internal services stay direct: mount a transport without a proxy
    "all://*.internal.example": httpx.HTTPTransport(),
}

with httpx.Client(mounts=mounts) as client:
    print(client.get("https://api.ipify.org?format=json").json())

より具体的なパターンが一般的なパターンよりも優先されるため、all://*.internal.exampleというマウントは、そのドメインについてはプロキシをバイパスし、それ以外はすべてルーティングされます。

URL での認証

Geonode の認証情報はプレーンテキストのユーザー名とパスワードであるため、http://user:pass@host:port という形式のユーザー情報だけで十分です。実用上の注意点として2点あります。まず、パスワードに @、:、/ などの文字が含まれている場合は、URL を構築する前に urllib.parse.quote のようにパーセントエンコードしてください。次に、機密情報をソース管理から除外するために、環境変数から読み込んでください：

import os
from urllib.parse import quote

import httpx

user = os.environ["GEONODE_USER"]
password = quote(os.environ["GEONODE_PASS"], safe="")
proxy = f"http://{user}:{password}@premium-residential.geonode.com:9000"

client = httpx.Client(proxy=proxy)

これ以降のすべての内容は、URLではなくユーザー名を拡張することでターゲティングやセッションのオプションを追加するため、同じパターンが引き続き機能します。

`Client` と `AsyncClient` の同期

非同期クライアントも、同じ proxy= キーワードを使用します。唯一の違いが現れるのは、mountsに移行したときです。AsyncClientにはhttpx.AsyncHTTPTransportが必要であり、httpx.HTTPTransportは必要ありません。同期トランスポートを非同期クライアントに渡すと、最初のリクエストで例外が発生します。

import asyncio
import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

async def fetch(client: httpx.AsyncClient, url: str) -> str:
    r = await client.get(url)
    r.raise_for_status()
    return r.json()["ip"]

async def main() -> None:
    transport = httpx.AsyncHTTPTransport(proxy=PROXY, retries=2)
    async with httpx.AsyncClient(
        mounts={"all://": transport}, timeout=httpx.Timeout(10.0, connect=5.0)
    ) as client:
        urls = ["https://api.ipify.org?format=json"] * 5
        ips = await asyncio.gather(*(fetch(client, u) for u in urls))
        print(ips)

asyncio.run(main())

ポート 9000 は接続ごとにローテーションされるため、5 つの同時リクエストでは通常、いくつかの異なる IP アドレスが表示されます。これはローテーションプールにおける想定通りの動作です。IP アドレスを一致させる必要がある場合は、スティッキーセッション（後述）を使用してください。

SOCKS5 ～ `httpx[socks]`

httpxプロキシの設定におけるSOCKSのサポートはオプションです。pip install "httpx[socks]"をインストールしてsocksioパッケージを追加し、その後スキームを切り替えてください。 Geonodeに含まれるSOCKS5のローテーションポートは11000です：

import httpx

PROXY = "socks5://USERNAME:PASSWORD@premium-residential.geonode.com:11000"

with httpx.Client(proxy=PROXY) as client:
    print(client.get("https://api.ipify.org?format=json").json())

追加のパッケージをインストールしていない場合、httpxはsocks5://プロキシURLを検出するとすぐに、socksioに言及したImportErrorエラーを発生させます。通常のスクレイピングでは、407のレスポンスが読みやすいため、HTTPポートの方がデバッグが簡単です。

環境変数と`trust_env`

デフォルトの trust_env=True では、httpx は HTTP_PROXY、HTTPS_PROXY、ALL_PROXY、および NO_PROXY を読み取り、それらを自動的に適用します。これはワークステーション上では便利ですが、CI環境では混乱の原因となります。CI環境では、環境内の企業プロキシがルーティングを黙って書き換えてしまうためです。コード内でプロキシを設定する場合は、環境変数の参照を無効にして、どこでも同じ動作になるようにすることを検討してください：

import httpx

client = httpx.Client(
    proxy="http://USERNAME:PASSWORD@premium-residential.geonode.com:9000",
    trust_env=False,
)

明示的な proxy= はすでに環境設定よりも優先されます。また、trust_env=False を設定すると、httpx が SSL_CERT_FILE を読み取るのを阻止するため、意図的に設定してください。

タイムアウトとトランスポートレベルのリトライ

プロキシはホップ数を増やすため、接続時間が長くなり、デフォルトの5秒のタイムアウトは一般家庭向けエグジットでは厳しい場合があります。httpx.Timeoutを使用すると、接続制限を個別に設定できる総予算を設定でき、HTTPTransport(retries=...)はリクエストが送信される前に、失敗した接続試行を再試行します。

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

transport = httpx.HTTPTransport(proxy=PROXY, retries=2)
timeout = httpx.Timeout(15.0, connect=8.0)

with httpx.Client(mounts={"all://": transport}, timeout=timeout) as client:
    r = client.get("https://api.ipify.org?format=json")
    print(r.json())

トランスポート再試行は、接続確立中の ConnectError および同様のエラーをカバーします。ステータスコードに基づく再試行は行われず、バイトが送信された後の再試行も行われないため、ゲートウェイからの 502 やターゲットサイトからの 429 は、依然としてコードに到達します。これらについては、独自の小さなループで処理し、その際にセッション ID を変更するかどうかを決定することもできます。

Geonode ユーザー名グラマーを使用したスティッキーセッション

ログイン、ページネーション、ダウンロードなどの多段階フローにおいて、接続ごとにIPアドレスをローテーションさせるのは適切ではありません。こうしたケースでは、httpxプロキシはリクエスト間で同じ出口IPを維持する必要があります。 Geonodeはこれをユーザー名にエンコードし、HTTP(S)スティッキーポート10000で提供します：

USER[-type-residential|datacenter|mix][-country-xx][-state-name|-city-name][-asn-N][-session-XXXXXXXX][-lifetime-SECONDS]

セッションIDは、ユーザーが生成する8文字の英数字です。有効期間は180秒から86400秒（24時間）です。州と都市を組み合わせることはできず、asnでは国の指定が必要です。

import secrets
import httpx

def sticky_client(country: str = "us", lifetime: int = 600) -> httpx.Client:
    session_id = secrets.token_hex(4)  # 8 hex characters
    username = f"USERNAME-country-{country}-session-{session_id}-lifetime-{lifetime}"
    proxy = f"http://{username}:PASSWORD@premium-residential.geonode.com:10000"
    return httpx.Client(proxy=proxy, timeout=httpx.Timeout(15.0, connect=8.0))

with sticky_client() as client:
    first = client.get("https://api.ipify.org?format=json").json()["ip"]
    second = client.get("https://api.ipify.org?format=json").json()["ip"]
    print(first == second, first)

新しい session_id は新しい IP を意味するため、「手動」でのローテーションは、単に新しいクライアントを作成することになります。スティッキーの有効期間は、ジョブの要件に合わせてできるだけ短く設定してください。

出口IPの確認

上記のすべての例ではhttps://api.ipify.org?format=jsonにリクエストが送信され、どのアドレスからアクセスしたかに関わらず、{"ip": "..."}が返されます。これを一度、直接呼び出し（プロキシなし）と比較すれば、プロキシが経路上にあることを証明する2行の簡易テストが完成します。他のデバッグを行う前に、まずこのテストを行ってください。多くの「httpxプロキシが機能しない」という報告は、環境変数やマウントパターンによってプロキシがバイパスされていたため、実際にはプロキシが使用されていなかったリクエストによるものです。

407 `ProxyError` および `ConnectError` の対処法

httpx プロキシの問題のほぼすべては、3 つの障害に起因しています。以下の表では、各障害の原因と対処法をまとめています。

症状	例外	一般的な原因	対処法
レスポンスステータス 407	`httpx.ProxyError`（HTTPS ターゲットのトンネル設定中に発生）または HTTP ターゲットに対する単純な 407 レスポンス	認証情報が間違っている、またはゲートウェイが拒否するユーザー名オプション	ユーザー名の構文を確認し、パスワードをパーセントエンコードする
トンネル拒否、プロキシからの 502/503	`httpx.ProxyError`	ゲートウェイがトンネルを開けなかったか、プロトコルに対してポートが間違っている	HTTP(S) の場合は 9000/10000、 SOCKS5の場合は11000/12000であることを確認してください
プロキシホストに接続できません	`httpx.ConnectError`	DNS、ファイアウォール、またはホスト名の誤り	ホストを手動で解決してください；`connect`のタイムアウトを延長してください
リクエストがハングし、その後失敗する	`httpx.ConnectTimeout` / `httpx.ReadTimeout`	住宅用回線の出口が遅い、またはターゲットがブロックされている	タイムアウトを延長し、新しいセッション ID で再試行

import httpx

PROXY = "http://USERNAME:PASSWORD@premium-residential.geonode.com:9000"

def get_via_proxy(url: str) -> httpx.Response | None:
    try:
        with httpx.Client(proxy=PROXY, timeout=httpx.Timeout(15.0, connect=8.0)) as client:
            r = client.get(url)
            if r.status_code == 407:
                raise RuntimeError("proxy rejected credentials (407)")
            return r
    except httpx.ProxyError as e:
        print("proxy refused tunnel:", e)
    except httpx.ConnectError as e:
        print("cannot connect to proxy host:", e)
    except httpx.TimeoutException as e:
        print("timed out:", type(e).__name__)
    return None

resp = get_via_proxy("https://api.ipify.org?format=json")
print(resp.json() if resp else "failed")

すべての httpx 例外は httpx.HTTPError を継承するため、ログ記録には単一の広範な except で対応できますが、ProxyError と ConnectError を区別することで、認証情報を確認すべきか、ネットワークを確認すべきかが判断できます。

よくある間違い

プロキシ URL での `https://` の使用

前述の通り、プロキシ URL スキームはプロキシへのホップを表します。ターゲットが https:// の場合でも、HTTP(S) ポートには http:// を使用してください。 https://は、プロキシに対して明示的にTLSを通知するゲートウェイ用に予約してください。

`AsyncClient`に`HTTPTransport`を混在させる場合

AsyncClient上のmountsは、AsyncHTTPTransportを使用する必要があります。同期トランスポートは構築時には受け入れられますが、リクエスト時には役に立たないエラーが発生して失敗するため、リクエストを送信しないテストではこの点を見落としがちです。

ローテーションポートからの IP アドレスが 1 つであることを想定

ポート 9000 は接続ごとにローテーションします。 2つのリクエストが連続して同じIPを共有する必要がある場合は、ユーザー名に-session- IDを含むポート10000を使用してください。Clientを再利用しても、プール内の接続が置き換えられる可能性があるため、それだけではIPを固定できません。

環境変数がコードを上書きするのを防ぐ

Docker イメージや CI ランナー内に HTTPS_PROXY が混入していると、直接接続だと思っていたトラフィックがルーティングされてしまいます。また、NO_PROXY を使用すると、プロキシ経由になるはずのホストが除外される可能性があります。プロキシルーティングを確定的にする必要がある場合は、trust_env=False を設定してください。

次のステップ

上記のクライアントが住宅用IPを返し、リクエスト間でスティッキーセッションが維持されるようになれば、残る変数はその背後にあるプロキシプールです。Geonodeの住宅用プロキシは、このガイドで使用されるエンドポイントを公開しています： 250万のIPがオンラインで利用可能、195カ国以上で毎月3,000万～5,000万のユニークIPがローテーションされ、ユーザー名の構文による国・州・都市・ASNのターゲティングが可能で、スティッキーセッションは最大24時間維持されます。価格はボリュームディスカウントで1GBあたり0.27ドルから、エントリープランは10GBで月額7.92ドル、3日間のトライアルは5ドルです。これを基盤としてクローラーを構築する場合は、ウェブスクレイピングのユースケースでは、プロキシとリトライロジックを組み合わせる方法について解説しており、ローテーション型データセンタープロキシは、データセンターのIP範囲をブロックしないターゲットに対しては、より安価な選択肢となります。

よくある質問

すべてのリクエストに対してhttpxプロキシを設定するにはどうすればよいですか？

クライアントのコンストラクタ（httpx.Client(proxy="http://user:pass@host:port")）にプロキシのURLを渡してください。そのクライアントを介して行われるすべてのリクエストは、http://およびhttps://のいずれのターゲットに対しても、プロキシを経由してルーティングされます。このキーワードは、httpx.AsyncClient や httpx.get() などのトップレベルヘルパーでも同様に機能します。URL に含まれる認証情報は、自動的に Proxy-Authorization として送信されます。

httpx におけるプロキシとマウントの違いは何ですか？

proxyは、すべての通信を1つのプロキシ経由でルーティングするショートカットです。mountsは、"http://"や"https://"などのURLパターンをトランスポートにマッピングし、各トランスポートは独自のプロキシ、再試行回数、制限を設定できます。スキームごとやホストごとに異なるプロキシが必要な場合、あるいはプロキシを設定しないトランスポートをマウントして特定の宛先に対してプロキシをバイパスしたい場合は、mountsを使用してください。

プロキシURLがhttps://?で始まる場合、なぜhttps://のマウントが失敗するのですか？

プロキシ URL のスキームは、httpx がターゲットではなくプロキシとどのように通信するかを記述するものです。https:// のターゲットの場合、httpx はプロキシへのプレーン HTTP 接続上で CONNECT トンネルを開くため、プロキシ URL には通常 http:// を使用する必要があります。 https:// を使用すると、httpx はプロキシ自体と TLS をネゴシエートするよう求められますが、ほとんどのプロキシエンドポイントはこれに対応していません。

httpxはSOCKS5プロキシをサポートしていますか？

はい。ただし、pip install httpx[socks]を含むオプションの依存関係をインストールした後でのみサポートされます。これによりsocksioが導入されます。その後、HTTPプロキシの場合とまったく同じようにproxy="socks5://user:pass@host:port"を指定してください。 Geonode を使用する場合、SOCKS5 のローテーションポートは 11000、SOCKS5 のスティッキーポートは 12000 です。HTTP(S) ポートは SOCKS ハンドシェイクを受け付けません。

プロキシ接続に失敗した場合、httpxはリクエストを自動的に再試行できますか？

トランスポートレベルでのみ可能です。httpx.HTTPTransport(retries=2)は、プロキシへの接続エラーを含む接続失敗の試行を再試行しますが、407や502などの応答がすでに返されているリクエストの再試行は決して行いません。 HTTPステータスコードに基づく再試行は、セッションIDを切り替えたり、意図的にバックオフを行ったりできる独自のループ内で行う必要があります。

httpxプロキシの設定において、一般家庭用IPアドレスでできることには制限がありますか？

プロキシはあなたの出口IPを変更しますが、サイトが許可する範囲を変更するわけではありません。robots.txt、利用規約、レート制限を遵守し、同時接続数を控えめに保ち、処理する根拠のない個人データの収集は避けてください。また、一般家庭用セッションには24時間の上限があり、それを過ぎるとIPが解放されるため、新しいセッションを開く必要があります。

PythonでのHTTPXプロキシ：設定、認証、非同期リクエスト

httpxプロキシの設定方法：proxyとmounts