Pythonでレジデンシャルプロキシを使う方法(requests、httpx、Scrapy)

Pythonにおけるレジデンシャルプロキシの実践ガイド:認証、ローテーティングプロキシと固定IPの違い、ジオターゲティング、リトライについて、requests、httpx、Scrapyの例を交えて解説します。

residential proxyをPythonスクリプトに組み込むのは、一度やり方を見れば5分で済む作業だ。詰まるのはコードそのものではなく、誰も書き残さない細部だ。認証情報がどうターゲティングを表現するか、IPをローテーションすべきか固定すべきか、プロキシ特有のエラーをどう処理するか、そしてrequests、httpx、Scrapyの間にある小さなライブラリの違いなどである。

これは筆者が欲しかったガイドだ。実際に動くコピペ可能な例を、よく使われる3つのライブラリ向けに用意し、さらに動くスニペットを本番環境で生き残るスクレイパーへと変える部分も収録している。

以下はすべて、Shifterのレジデンシャルゲートウェイを使用している。エンドポイントはp.shifter.io:443の1つだけで、ターゲティングはすべてユーザー名にエンコードされる。別のプロバイダーを使っている場合も構造は同じで、ホストと認証情報の形式を差し替えるだけでよい。

まず理解すべきこと一つ

レジデンシャルゲートウェイでは、プロキシのユーザー名が認証情報とターゲティングの両方を運ぶ。国やセッションを切り替えるためにエンドポイントを変えるのではなく、ユーザー名の文字列を変える。ユーザー名は次のような形になる。

customer-USERNAME-country-us-sid-abc123-ttl-600

左から右に読んでいく。アカウントID、続いてフラグだ。country-usは米国をターゲットにする。sid-abc123はスティッキーセッションを固定する。ttl-600はそのIPを600秒間保持する。sid/ttlを外すと、すべてのリクエストで新しいIPにローテーションする。パスワードは固定だ。これが全体のメンタルモデルであり、残りはこの文字列を各ライブラリのプロキシ設定に差し込むだけだ。

認証情報は環境変数に保持し、ハードコードしないこと。

import os
USER = os.environ["SHIFTER_USER"]   # your account username
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

requests: 6行版

requestsはproxies辞書を受け取る。httpキーとhttpsキーは両方とも同じhttp://プロキシURLを指す。これは正しい挙動で、requestsはCONNECTを介してHTTPSをそのプロキシ経由でトンネリングする。

import os, requests

USER = os.environ["SHIFTER_USER"]
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

def proxy(country="us"):
    url = f"http://{USER}-country-{country}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

r = requests.get("https://api.ipify.org?format=json", proxies=proxy("us"), timeout=30)
print(r.json())   # {'ip': '<a US residential IP>'}

2回実行すると2つの異なるIPが返ってくる。なぜならsidがなければ、すべてのリクエストがローテーションするからだ。これがデフォルトの動作であり、多くのスクレイピング用途ではこれがまさに望ましい挙動になる。

ローテーションとスティッキー、コードで見る

これは混乱しがちな区別なので、具体的に示す。（概念的な説明はsticky vs rotating residential proxiesを参照。）

ローテーティング（リクエストごとに新しいIP）はデフォルトであり、sidを省略するだけでよい。

for _ in range(3):
    r = requests.get("https://api.ipify.org", proxies=proxy("us"), timeout=30)
    print(r.text)   # three different IPs

スティッキー（複数リクエストで同一IP）はユーザー名にセッションIDとTTLが必要だ。ログインとその後続ページのように、複数のリクエストが1人のユーザーに見える必要があるフローで使う。

def sticky_proxy(country="us", session="s1", ttl=600):
    url = f"http://{USER}-country-{country}-sid-{session}-ttl-{ttl}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

s = requests.Session()
p = sticky_proxy(session="checkout-42", ttl=600)
for path in ("/login", "/cart", "/checkout"):
    r = s.get(f"https://shop.example{path}", proxies=p, timeout=30)
    # all three requests share one IP for up to 600s

セッションIDは自分で選ぶ。論理的なセッションごとに一意な文字列であればよい。同じ文字列はTTLが切れるまで同じIPを返し、その後は他のフラグを保持したままローテーションする。

地域ターゲティング

ターゲティングがユーザー名に存在するため、地域指定も単なるフラグの一つだ。国指定が最も一般的だが、州、都市、ASNも同様に機能する。

def geo_proxy(country, city=None):
    parts = [USER, "country", country]
    if city:
        parts += ["city", city.lower().replace(" ", "_")]
    url = f"http://{'-'.join(parts)}:{PASS}@{GATEWAY}"
    return {"http": url, "https": url}

requests.get("https://example.com", proxies=geo_proxy("de"))              # Germany
requests.get("https://example.com", proxies=geo_proxy("us", "new york"))  # New York City

都市名はスペースをアンダースコアで置き換える（new_york）。フラグは自由に組み合わせられ、ゲートウェイにとって順序は問題にならない。

httpx: 同じ考え方、非同期対応

非同期やHTTP/2を使いたい場合、httpxは現代的な選択肢だ。プロキシはクライアントに設定する。パラメータ名は現行のhttpxではproxy=（単数形）であることに注意。古いバージョンではproxies=だった。

import os, httpx

USER = os.environ["SHIFTER_USER"]
PASS = os.environ["SHIFTER_PASS"]
GATEWAY = "p.shifter.io:443"

def proxy_url(country="us"):
    return f"http://{USER}-country-{country}:{PASS}@{GATEWAY}"

# Sync
with httpx.Client(proxy=proxy_url("us"), timeout=30) as client:
    print(client.get("https://api.ipify.org").text)

httpxが真価を発揮するのは非同期版だ。多数のリクエストを並行して発行し、それぞれが新しいローテーティングIPを経由する。

import asyncio, httpx

async def fetch(client, url):
    r = await client.get(url, timeout=30)
    return r.status_code, r.text[:80]

async def main(urls):
    async with httpx.AsyncClient(proxy=proxy_url("us")) as client:
        return await asyncio.gather(*(fetch(client, u) for u in urls))

urls = ["https://api.ipify.org"] * 10
print(asyncio.run(main(urls)))   # 10 concurrent requests, rotating IPs

これで10並行のレジデンシャルリクエストが十数行で実現できる。並行数には注意が必要だ。多ければ必ず速くなるわけではなく、1つのターゲットに対して多数のIPから一気にアクセスすると、行動検知に引っかかることもある。

Scrapy: プロキシミドルウェア

大規模なクロールにはScrapyが重量級の選択肢になる。プロキシを設定する適切な方法は、リクエストごとにrequest.meta["proxy"]を設定することだ。Scrapy組み込みのHttpProxyMiddlewareがこれを読み取り、URL内の認証情報からProxy-Authorizationヘッダーを処理する。

地域をローテーションし、リクエストごとに新しいIPを与える小さなミドルウェアの例。

import os

class ResidentialProxyMiddleware:
    def __init__(self):
        self.user = os.environ["SHIFTER_USER"]
        self.password = os.environ["SHIFTER_PASS"]
        self.gateway = "p.shifter.io:443"

    def process_request(self, request, spider):
        country = request.meta.get("country", "us")
        request.meta["proxy"] = (
            f"http://{self.user}-country-{country}:"
            f"{self.password}@{self.gateway}"
        )

settings.pyで有効化する（デフォルトのプロキシミドルウェアより先に実行される必要がある）。

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.ResidentialProxyMiddleware": 350,
    "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 400,
}

これで任意のスパイダーがレジデンシャルIPを経由するようになり、Request(url, meta={"country": "gb"})でリクエストごとにターゲットを指定できる。スティッキーセッションの場合は、requestsの例とまったく同じ方法でsid/ttlを含むユーザー名を組み立て、meta["proxy"]に設定する。

実際に発生するエラーへの対処

プロキシのエラーを無視するスクレイパーは、デモでは動くが夜間には落ちる。実際に遭遇する3つのエラーを挙げる。

407 Proxy Authentication Required、ユーザー名またはパスワードが間違っている、あるいは認識されないフラグ（countryやasnのタイプミス）が原因。認証情報の文字列を修正すること。リトライしても解決しない。
429 Too Many Requests、ターゲット側がレート制限をかけている。バックオフして速度を落とし、IPをローテーションする（スティッキーでなければゲートウェイがリクエストごとに既に行っている）。
502 Bad Gateway、指定したフィルターに合致するIPが現在存在しない（通常は国+都市+ASNのように地域指定が過度に厳しい場合）。フラグを緩めて再試行する。

バックオフし、きれいに諦める最小限のリトライラッパー。

import time, requests

def get_with_retry(url, proxies, tries=4):
    for attempt in range(tries):
        try:
            r = requests.get(url, proxies=proxies, timeout=30)
            if r.status_code in (429, 502, 503):
                raise requests.exceptions.RequestException(f"status {r.status_code}")
            r.raise_for_status()
            return r
        except requests.exceptions.RequestException as e:
            if attempt == tries - 1:
                raise
            sleep = 2 ** attempt           # 1s, 2s, 4s
            print(f"retry {attempt+1}: {e}; sleeping {sleep}s")
            time.sleep(sleep)

指数バックオフとリクエストごとのローテーションの組み合わせは、一時的な失敗の大部分を処理できる。断続的ではなく常にブロックされる場合は、リトライの問題ではなく、IP品質やリクエストの挙動の問題だ。詳しくはhow to avoid getting blocked when scrapingを参照してほしい。

実際に動作しているかを確認する

設定を信頼する前に、2つのことを確認しよう。IPが変わること（ローテーション）と、正しい国であること（地域指定）だ。簡単な確認方法は次の通り。

import requests

r = requests.get("http://ip-api.com/json", proxies=proxy("de"), timeout=30)
data = r.json()
print(data["query"], data["countryCode"])   # expect a DE IP

sidを付けずに数回呼び出すと、すべてDEでありながら異なるIPが返るはずだ。国が間違っている場合は、フラグの表記を確認すること。IPがまったく変わらない場合は、ユーザー名に誤ってsidが残っている。

SOCKS5についての補足

上記はすべてHTTPプロキシを使用しており、これはウェブスクレイピングにおける正しいデフォルトだ。ワークロードがSOCKS5を必要とする場合（HTTP以外のトラフィック、またはそれを前提とするツール）、同じゲートウェイがSOCKS5にも対応している。スキームをsocks5h://に変更し、requests[socks]をインストールする（またはhttpxのSOCKS拡張を使う）。トレードオフについてはHTTP vs SOCKS5 proxiesを参照してほしい。通常のスクレイピングではHTTPを使い続けるのがよい。

FAQ

国ごとに異なるエンドポイントが必要ですか？ 不要だ。p.shifter.io:443という1つのエンドポイントだけですべてに対応する。国、都市、セッションはすべてホストではなくユーザー名文字列で変更する。これがゲートウェイモデルの核心だ。

requestsでhttpとhttpsの両方のキーがhttp://のURLに設定されているのはなぜですか？ requestsはCONNECTトンネルを介してHTTPプロキシ経由でHTTPSを送るためだ。プロキシURLのスキームはプロキシとの通信方法（HTTP）を表しており、ターゲットとの通信方法ではない。これは正しく標準的な挙動であり、httpsキーをhttps://に設定してはいけない。

リクエストごとにIPをローテーションするにはどうすればよいですか？ ユーザー名からsidを省略する。セッションIDがなければ、ゲートウェイは自動的にリクエストごとに新しいIPを渡す。プロキシリストを自分で管理する必要はなく、プールのローテーションはサーバー側で行われる。

複数ステップのフローで同じIPを維持するにはどうすればよいですか？ ユーザー名にsid-<your-id>-ttl-<seconds>を追加し、フロー内の各リクエストで同じものを再利用する。同じIDであれば、TTLが切れるまで同じIPになる。

リクエストが遅いです。プロキシが原因ですか？ 通常は違う。レジデンシャルIPは直接接続に比べて多少のレイテンシを追加するが、規模が大きくなったときの遅さは、並行数が高すぎること、接続が再利用されていないこと（Session/Clientを使うこと）、あるいはターゲット自体が遅いことに起因する場合が多い。プロキシを責める前にプロファイリングすること。

aiohttp、urllib3、pycurlでも使えますか？ 使える。認証済みのhttp://user:pass@host:port形式のプロキシURLを受け付けるクライアントであれば動作し、認証情報にエンコードされたターゲティングの仕組みは同一だ。requests、httpx、Scrapyは単に最も一般的な3つに過ぎない。

まとめ

パターンはどのライブラリでも同じだ。http://USER-flags:PASS@p.shifter.io:443というURLを組み立て、プロキシ設定に渡し、ローテーションするならsidを省略し、固定するなら追加する。地域指定はフラグの一つに過ぎず、エラー処理は小さなリトライループで済み、並行処理はクライアントが既に持っている仕組みをそのまま使うだけだ。

上記のスニペットから始めて、レジデンシャルゲートウェイを指定すれば、午後のうちに本番仕様のプロキシコードが手に入る。プランとGB単価はpricing pageにあり、フラグの完全なリファレンスはgateway docsにある。