Pythonのrequestsとaiohttpでのプロキシ設定：コード例付き完全ガイド

パーサーの開発、データ収集の自動化、またはPythonからのWebサービスのテストを行う際には、プロキシサーバーを使用する必要があることがよくあります。requestsおよびaiohttpライブラリは、プロキシを扱うための柔軟なメカニズムを提供しますが、その設定には重要なニュアンスがあります。このガイドでは、同期的および非同期的アプローチを詳しく説明し、HTTPおよびSOCKS5プロキシの例を示し、IPのローテーションとエラーハンドリングを考察します。

requestsにおけるプロキシの基本設定

requestsライブラリは、PythonにおけるHTTPリクエストの標準です。プロキシの設定は、プロキシサーバーのプロトコルとアドレスを含む辞書を受け取るproxiesパラメータを介して行われます。

HTTPプロキシの最も簡単な例：

import requests

# プロキシの設定
proxies = {
    'http': 'http://123.45.67.89:8080',
    'https': 'http://123.45.67.89:8080'
}

# プロキシを介してリクエストを実行
response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())  # {'origin': '123.45.67.89'}

注意：HTTPSリクエストの場合、プロキシの値にはhttp://プロトコルも指定する必要があります（https://ではありません）。これは、プロキシサーバーとの接続がHTTPで確立され、その後CONNECTメソッドを介してHTTPSトラフィックのトンネルが作成されるためです。

環境変数の使用：

requestsライブラリは、環境変数HTTP_PROXYおよびHTTPS_PROXYから自動的にプロキシを読み取ります：

import os
import requests

# 環境変数を介して設定
os.environ['HTTP_PROXY'] = 'http://123.45.67.89:8080'
os.environ['HTTPS_PROXY'] = 'http://123.45.67.89:8080'

# プロキシが自動的に適用されます
response = requests.get('https://httpbin.org/ip')
print(response.json())

このアプローチは、コンテナ化（Docker）やシステムレベルでプロキシが設定される場合に便利です。ただし、柔軟性を考慮すると、proxiesパラメータを明示的に渡すことをお勧めします。

requestsにおける認証とSOCKS5

ほとんどの商用プロキシサービスは、ユーザー名とパスワードによる認証を要求します。requestsでは、資格情報を含むURL形式を介してこれを実現します。

認証付きHTTPプロキシ：

import requests

# フォーマット: http://username:password@host:port
proxies = {
    'http': 'http://user123:pass456@proxy.example.com:8080',
    'https': 'http://user123:pass456@proxy.example.com:8080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(response.json())

SOCKS5プロキシの設定：

SOCKS5を使用するには、追加のライブラリrequests[socks]またはPySocksが必要です。インストール方法：

pip install requests[socks]

SOCKS5の使用例：

import requests

# 認証なしのSOCKS5
proxies = {
    'http': 'socks5://123.45.67.89:1080',
    'https': 'socks5://123.45.67.89:1080'
}

# 認証付きSOCKS5
proxies_auth = {
    'http': 'socks5://user:pass@123.45.67.89:1080',
    'https': 'socks5://user:pass@123.45.67.89:1080'
}

response = requests.get('https://httpbin.org/ip', proxies=proxies_auth)
print(response.json())

SOCKS5プロキシは、レジデンシャルプロキシを使用する際に特に便利です。このプロトコルは、トラフィックのより信頼性の高いトンネリングを提供し、UDPをサポートします（いくつかのアプリケーションに必要です）。

requestsにおけるプロキシのローテーション

大量のデータをパースする際に、1つのIPアドレスを使用するとブロックされることがあります。プロキシのローテーションは、負荷を分散し、レート制限を回避するためのIPの循環的な変更です。

リストを使用した簡単なローテーション：

import requests
import itertools

# プロキシサーバーのリスト
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

# 無限イテレータの作成
proxy_pool = itertools.cycle(proxy_list)

# ローテーションを使用してリクエストを実行
for i in range(10):
    proxy = next(proxy_pool)
    proxies = {'http': proxy, 'https': proxy}
    
    try:
        response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
        print(f"リクエスト {i+1}: IP = {response.json()['origin']}")
    except Exception as e:
        print(f"プロキシ {proxy} に関するエラー: {e}")

クッキーを保持するためのセッションを使用したローテーション：

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list):
        self.proxy_pool = cycle(proxy_list)
        self.session = requests.Session()
    
    def get(self, url, **kwargs):
        proxy = next(self.proxy_pool)
        self.session.proxies = {'http': proxy, 'https': proxy}
        return self.session.get(url, **kwargs)

# 使用例
proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
]

rotator = ProxyRotator(proxy_list)

for i in range(5):
    response = rotator.get('https://httpbin.org/ip', timeout=5)
    print(f"リクエスト {i+1}: {response.json()['origin']}")

予測不可能性のためのランダムローテーション：

import requests
import random

proxy_list = [
    'http://user:pass@proxy1.example.com:8080',
    'http://user:pass@proxy2.example.com:8080',
    'http://user:pass@proxy3.example.com:8080',
]

def get_random_proxy():
    proxy = random.choice(proxy_list)
    return {'http': proxy, 'https': proxy}

# 各リクエストにランダムプロキシを使用
for i in range(5):
    response = requests.get('https://httpbin.org/ip', proxies=get_random_proxy(), timeout=5)
    print(f"リクエスト {i+1}: {response.json()['origin']}")

ランダムローテーションは、リクエストパターンを追跡するサイトで効果的です。IPの順次変更は疑わしく見える可能性がありますが、ランダムな選択は異なるユーザーの行動を模倣します。

aiohttpにおけるプロキシの設定

aiohttpライブラリは、非同期HTTPリクエスト用に設計されており、高負荷のパーサーには重要です。プロキシの設定はrequestsとは異なり、単数形のproxyパラメータを使用します。

HTTPプロキシの基本例：

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = 'http://123.45.67.89:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            data = await response.json()
            print(data)

# 実行
asyncio.run(fetch_with_proxy())

認証付きプロキシ：

aiohttpでは、認証はaiohttp.BasicAuthオブジェクトを介して、またはURLに直接渡されます：

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    # オプション1: URLに資格情報を含める
    proxy = 'http://user123:pass456@proxy.example.com:8080'
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', proxy=proxy) as response:
            print(await response.json())

# オプション2: BasicAuthを使用（いくつかのプロキシ用）
async def fetch_with_basic_auth():
    proxy = 'http://proxy.example.com:8080'
    proxy_auth = aiohttp.BasicAuth('user123', 'pass456')
    
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/ip', 
                                proxy=proxy, 
                                proxy_auth=proxy_auth) as response:
            print(await response.json())

asyncio.run(fetch_with_auth_proxy())

aiohttpにおけるSOCKS5：

SOCKS5を使用するには、aiohttp-socksライブラリが必要です：

pip install aiohttp-socks

import asyncio
from aiohttp_socks import ProxyConnector
import aiohttp

async def fetch_with_socks5():
    connector = ProxyConnector.from_url('socks5://user:pass@123.45.67.89:1080')
    
    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get('https://httpbin.org/ip') as response:
            print(await response.json())

asyncio.run(fetch_with_socks5())

モバイルプロキシを使用してソーシャルメディアやマーケットプレイスをパースする際には、aiohttpを使用することをお勧めします。非同期性により、数百のリクエストを並行して処理することができ、実行スレッドがブロックされることはありません。

非同期ローテーションとプロキシプール

高負荷のパーサーには、障害処理と非稼働IPの自動交換を伴う効果的なプロキシローテーションが必要です。aiohttpのための高度なパターンを考察します。

プロキシプールを管理するクラス：

import aiohttp
import asyncio
from itertools import cycle
from typing import List, Optional

class ProxyPool:
    def __init__(self, proxy_list: List[str]):
        self.proxy_list = proxy_list
        self.proxy_cycle = cycle(proxy_list)
        self.failed_proxies = set()
    
    def get_next_proxy(self) -> Optional[str]:
        """次の作動するプロキシを取得"""
        for _ in range(len(self.proxy_list)):
            proxy = next(self.proxy_cycle)
            if proxy not in self.failed_proxies:
                return proxy
        return None  # すべてのプロキシが利用できない
    
    def mark_failed(self, proxy: str):
        """プロキシを非稼働としてマーク"""
        self.failed_proxies.add(proxy)
        print(f"プロキシ {proxy} は利用できないとマークされました")
    
    async def fetch(self, session: aiohttp.ClientSession, url: str, **kwargs):
        """エラー時に自動的にプロキシを変更してリクエストを実行"""
        max_retries = 3
        
        for attempt in range(max_retries):
            proxy = self.get_next_proxy()
            if not proxy:
                raise Exception("すべてのプロキシが利用できません")
            
            try:
                async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10), **kwargs) as response:
                    return await response.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                print(f"プロキシ {proxy} に関するエラー: {e}")
                self.mark_failed(proxy)
                continue
        
        raise Exception(f"{max_retries} 回の試行の後にリクエストを実行できませんでした")

# 使用例
async def main():
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
        'http://user:pass@proxy3.example.com:8080',
    ]
    
    pool = ProxyPool(proxy_list)
    
    async with aiohttp.ClientSession() as session:
        # 自動ローテーションで10回リクエストを実行
        tasks = [pool.fetch(session, 'https://httpbin.org/ip') for _ in range(10)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"リクエスト {i+1} はエラーで終了しました: {result}")
            else:
                print(f"リクエスト {i+1}: IP = {result.get('origin')}")

asyncio.run(main())

同時実行制限を持つ並行処理：

import aiohttp
import asyncio
from itertools import cycle

async def fetch_url(session, url, proxy, semaphore):
    async with semaphore:  # 同時リクエストの制限
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as response:
                data = await response.json()
                return {'url': url, 'ip': data.get('origin'), 'status': response.status}
        except Exception as e:
            return {'url': url, 'error': str(e)}

async def main():
    urls = [f'https://httpbin.org/ip' for _ in range(50)]  # 50リクエスト
    proxy_list = [
        'http://user:pass@proxy1.example.com:8080',
        'http://user:pass@proxy2.example.com:8080',
    ]
    proxy_cycle = cycle(proxy_list)
    
    # 制限: 同時リクエストは10まで
    semaphore = asyncio.Semaphore(10)
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, url, next(proxy_cycle), semaphore)
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        
        # 結果の分析
        successful = [r for r in results if 'ip' in r]
        failed = [r for r in results if 'error' in r]
        
        print(f"成功したリクエスト: {len(successful)}")
        print(f"失敗したリクエスト: {len(failed)}")

asyncio.run(main())

asyncio.Semaphoreを使用することは、プロキシを使用する際に非常に重要です。1つのIPを介して同時接続が多すぎると、ターゲットサイトやプロキシプロバイダーによってブロックされる可能性があります。

エラーハンドリングとタイムアウト

プロキシを使用する際には、タイムアウト、接続の切断、プロキシサーバーの拒否など、エラーが増える可能性があります。エラーの正しい処理は、パーサーの安定性の鍵です。

プロキシ使用時の一般的なエラー：

requestsでのエラーハンドリング：

import requests
from requests.exceptions import ProxyError, ConnectTimeout, RequestException

def fetch_with_retry(url, proxies, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url, 
                proxies=proxies, 
                timeout=(5, 10),  # (接続タイムアウト, 読み取りタイムアウト)
                allow_redirects=True,
                verify=True  # SSL証明書の検証
            )
            response.raise_for_status()  # 4xx/5xxの場合は例外を発生させる
            return response.json()
            
        except ProxyError as e:
            print(f"試行 {attempt + 1}: プロキシが利用できません - {e}")
        except ConnectTimeout as e:
            print(f"試行 {attempt + 1}: 接続タイムアウト - {e}")
        except requests.exceptions.HTTPError as e:
            print(f"試行 {attempt + 1}: HTTPエラー {e.response.status_code}")
            if e.response.status_code == 407:  # プロキシ認証が必要
                print("プロキシ認証エラー！")
                break  # 認証エラーの場合は再試行しない
        except RequestException as e:
            print(f"試行 {attempt + 1}: 一般的なエラー - {e}")
        
        if attempt < max_retries - 1:
            print(f"2秒後に再試行...")
            import time
            time.sleep(2)
    
    raise Exception(f"{max_retries} 回の試行の後にリクエストを実行できませんでした")

# 使用例
proxies = {'http': 'http://user:pass@proxy.example.com:8080', 'https': 'http://user:pass@proxy.example.com:8080'}
try:
    data = fetch_with_retry('https://httpbin.org/ip', proxies)
    print(data)
except Exception as e:
    print(f"重大なエラー: {e}")

aiohttpでのエラーハンドリング：

import aiohttp
import asyncio
from aiohttp import ClientError, ClientProxyConnectionError

async def fetch_with_retry(session, url, proxy, max_retries=3):
    for attempt in range(max_retries):
        try:
            timeout = aiohttp.ClientTimeout(total=10, connect=5)
            async with session.get(url, proxy=proxy, timeout=timeout) as response:
                response.raise_for_status()
                return await response.json()
                
        except ClientProxyConnectionError as e:
            print(f"試行 {attempt + 1}: プロキシ接続エラー - {e}")
        except asyncio.TimeoutError:
            print(f"試行 {attempt + 1}: タイムアウト")
        except aiohttp.ClientHttpProxyError as e:
            print(f"試行 {attempt + 1}: プロキシのHTTPエラー - {e}")
            if e.status == 407:
                print("プロキシ認証エラー！")
                break
        except ClientError as e:
            print(f"試行 {attempt + 1}: クライアントの一般的なエラー - {e}")
        
        if attempt < max_retries - 1:
            await asyncio.sleep(2)
    
    raise Exception(f"{max_retries} 回の試行の後にリクエストを実行できませんでした")

async def main():
    proxy = 'http://user:pass@proxy.example.com:8080'
    async with aiohttp.ClientSession() as session:
        try:
            data = await fetch_with_retry(session, 'https://httpbin.org/ip', proxy)
            print(data)
        except Exception as e:
            print(f"重大なエラー: {e}")

asyncio.run(main())

タイムアウトの設定：

タイムアウトの正しい設定は安定性にとって重要です。推奨される値は次のとおりです：

• 接続タイムアウト： 5-10秒（プロキシとの接続を確立するための時間）
• 読み取りタイムアウト： 10-30秒（ターゲットサイトからの応答を受け取るための時間）
• 合計タイムアウト： 30-60秒（リクエストの合計時間）

遅いレジデンシャルプロキシの場合、接続ごとにタイムアウトを20-30秒に増やすことをお勧めします。実際のプロバイダーを介したルーティングには、より多くの時間がかかる可能性があります。

ベストプラクティスと最適化

プロキシを効果的に使用するには、ブロックを最小限に抑え、パフォーマンスを最大化するための一連のルールを遵守する必要があります。

1. 接続の再利用のためにSessionを使用：

# requests: SessionはTCP接続を再利用します
session = requests.Session()
session.proxies = {'http': proxy, 'https': proxy}

for url in urls:
    response = session.get(url)  # requests.get()よりも速いです

# aiohttp: 非同期性のためにSessionが必須です
async with aiohttp.ClientSession() as session:
    tasks = [session.get(url, proxy=proxy) for url in urls]
    await asyncio.gather(*tasks)

2. 現実的なUser-Agentとヘッダーの設定：

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.5',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

proxies = {'http': proxy, 'https': proxy}
response = requests.get('https://example.com', headers=headers, proxies=proxies)

3. レート制限（秒あたりのリクエスト数）の制限：

import time
import requests

class RateLimiter:
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
    
    def wait(self):
        elapsed = time.time() - self.last_request_time
        if elapsed < self.interval:
            time.sleep(self.interval - elapsed)
        self.last_request_time = time.time()

# 使用例: 秒あたり2リクエストを超えない
limiter = RateLimiter(2)
proxies = {'http': proxy, 'https': proxy}

for url in urls:
    limiter.wait()
    response = requests.get(url, proxies=proxies)

4. プロキシのロギングと監視：

import logging
from collections import defaultdict

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

class ProxyMonitor:
    def __init__(self):
        self.stats = defaultdict(lambda: {'success': 0, 'failed': 0, 'total_time': 0})
    
    def log_request(self, proxy, success, response_time):
        stats = self.stats[proxy]
        if success:
            stats['success'] += 1
        else:
            stats['failed'] += 1
        stats['total_time'] += response_time
        
        # 10リクエストごとにロギング
        total = stats['success'] + stats['failed']
        if total % 10 == 0:
            avg_time = stats['total_time'] / total
            success_rate = stats['success'] / total * 100
            logger.info(f"プロキシ {proxy}: {total} リクエスト, 成功率 {success_rate:.1f}%, 平均 {avg_time:.2f}s")

monitor = ProxyMonitor()

# リクエストコード内で
import time
start = time.time()
try:
    response = requests.get(url, proxies=proxies, timeout=10)
    monitor.log_request(proxy, True, time.time() - start)
except Exception as e:
    monitor.log_request(proxy, False, time.time() - start)
    logger.error(f"プロキシ {proxy} に関するエラー: {e}")

5. DNSキャッシュによる高速化：

# aiohttpでDNSキャッシュを使用
import aiohttp
from aiohttp.resolver import AsyncResolver

resolver = AsyncResolver(nameservers=['8.8.8.8', '8.8.4.4'])
connector = aiohttp.TCPConnector(resolver=resolver, ttl_dns_cache=300)

async with aiohttp.ClientSession(connector=connector) as session:
    # リクエストは5分間DNSキャッシュを使用します
    async with session.get(url, proxy=proxy) as response:
        data = await response.json()

6. CAPTCHAとブロックの処理：

プロキシに対するrequestsとaiohttpの比較

requestsとaiohttpの選択は、タスクとデータ量によります。主要な違いを考察します。

requestsを使用する場合：

• 単発タスクのための簡単なスクリプト
• プロトタイピングとテスト
• 小規模なリクエスト（1分あたり100まで）
• コードの簡潔さと可読性が重要な場合
• 同期ライブラリとの統合

aiohttpを使用する場合：

• 大量のデータをパースする（数千ページ）
• リアルタイムで多数のソースを監視
• 高負荷のAPIサービス
• 処理速度が重要な場合
• プロキシを介してWebSocketを使用する

パフォーマンスの実際の比較：

# テスト: プロキシを介して100リクエスト

# requests（同期） - 約50秒
import requests
import time

start = time.time()
proxies = {'http': proxy, 'https': proxy}
for i in range(100):
    response = requests.get('https://httpbin.org/ip', proxies=proxies)
print(f"requests: {time.time() - start:.2f} 秒")

# aiohttp（非同期） - 約5秒
import aiohttp
import asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as session:
        tasks = [
            session.get('https://httpbin.org/ip', proxy=proxy)
            for _ in range(100)
        ]
        await asyncio.gather(*tasks)

start = time.time()
asyncio.run(fetch_all())
print(f"aiohttp: {time.time() - start:.2f} 秒")

データセンターのプロキシを使用して高速でパースする場合、aiohttpはrequestsに比べて10-20倍の優位性を示します。これはリクエストの並行処理によるものです。

結論

Pythonにおけるrequestsおよびaiohttpを介したプロキシの設定は、パーサーの開発、データ収集の自動化、および地理的制限の回避にとって基本的なスキルです。requestsライブラリは、理解しやすい同期APIのおかげで簡単なスクリプトやプロトタイピングに適しており、aiohttpは非同期アーキテクチャを使用して数千のリクエストを処理する際に高いパフォーマンスを提供します。

Pythonでプロキシを効果的に使用するための重要なポイントは、エラーとタイムアウトの正しい処理、負荷分散のためのIPアドレスのローテーションの実装、接続の再利用のためのSessionの使用、現実的なヘッダーとUser-Agentの設定、プロキシサーバーのパフォーマンスの監視です。SOCKS5プロキシには、requests[socks]またはaiohttp-socksなどの追加ライブラリが必要です。

パースのためのプロキシの種類を選択する際には、タスクの特性を考慮してください。高負荷のパーサーには、高速なデータセンターのプロキシが適しており、厳しいアンチボットシステムを回避し、ソーシャルメディアを扱う場合には、実際の家庭ユーザーのIPを持つレジデンシャルプロキシが推奨されます。また、最大の匿名性とモバイルトラフィックの模倣が必要なタスクには、携帯電話のプロキシが最適です。

高性能なパーサーを開発したり、多くのソースからデータを自動的に収集したりする予定がある場合は、レジデンシャルプロキシを試してみることをお勧めします。これらは高い匿名性を提供し、ブロックのリスクを最小限に抑え、ほとんどの保護されたWebサービスで安定した動作を保証します。技術的なタスクで高速処理が必要な場合は、低遅延と高スループットを持つデータセンターのプロキシも適しています。