Python requests.exceptions.ConnectionError の原因と解決方法【API連携とネットワークトラブル対策】

requests.exceptions.ConnectionError とは

Pythonで外部APIと連携する際やWebスクレイピングを行う際に、requests.exceptions.ConnectionError に遭遇して困ったことはありませんか？このエラーは、名前の通りネットワーク接続に関する問題で、プログラムが指定されたサーバーに到達できなかったり、接続が予期せず切断されたりした場合に発生します。本記事では、このエラーの一般的な原因と、Python開発者が現場で使える実践的な解決策を詳しく解説します。

実行環境ごとのエラーメッセージ

環境	エラーメッセージ
Python 3.x (requests library)	requests.exceptions.ConnectionError: ('Connection aborted.', RemoteDisconnected('Remote end closed connection without response'))
Python 3.x (requests library, DNS failure)	requests.exceptions.ConnectionError: HTTPSConnectionPool(host='non-existent-domain.com', port=443): Max retries exceeded with url: / (Caused by NewConnectionError(': Failed to establish a new connection: [Errno -2] Name or service not known'))
Python 3.x (requests library, SSL error)	requests.exceptions.ConnectionError: HTTPSConnectionPool(host='expired.badssl.com', port=443): Max retries exceeded with url: / (Caused by SSLError(CertificateError("hostname 'expired.badssl.com' doesn't match 'badssl.com'")))

エラーの発生パターン

このエラーは主に以下のようなケースで発生します。

パターン1: 対象サーバーが起動していない、またはURLが間違っている

/** bad_code.py */
import requests

try:
    # 存在しないドメイン、またはポートが間違っている
    response = requests.get('http://non-existent-api.com:8001/data', timeout=5)
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
    # 出力例:
    # Connection Error: HTTPSConnectionPool(host='non-existent-api.com', port=8001): Max retries exceeded with url: /data (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x...>: Failed to establish a new connection: [Errno -2] Name or service not known'))

リクエスト先のサーバーがダウンしている、指定したURL（ドメイン、IPアドレス、ポート番号）が間違っている、またはDNS解決に失敗している場合に発生します。requestsライブラリはサーバーへのTCP接続を確立できず、このエラーを発生させます。

/** good_code.py */
import requests

try:
    # 正しいドメインとポート、サーバーが起動していることを確認
    # 例: 自分で立てたテスト用APIサーバーなど
    response = requests.get('http://httpbin.org/get', timeout=5)
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
except requests.exceptions.Timeout as e:
    print(f"Timeout Error: {e}")
except requests.exceptions.RequestException as e:
    print(f"Other Request Error: {e}")

パターン2: ネットワーク接続の問題（ファイアウォール、プロキシ、DNS）

/** bad_code.py */
import requests

# 会社のファイアウォールやプロキシ設定が正しくない状態で外部APIにアクセス
try:
    response = requests.get('https://api.external.com/items', timeout=10)
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
    # 出力例:
    # Connection Error: HTTPSConnectionPool(host='api.external.com', port=443): Max retries exceeded with url: /items (Caused by ProxyError('Cannot connect to proxy.', OSError('Tunnel connection failed: 407 Proxy Authentication Required')))

プログラムを実行している環境のファイアウォールが通信をブロックしている、プロキシ設定が誤っている、またはDNSサーバーが正しくドメインを解決できない場合に発生します。特に企業ネットワーク環境でよく見られます。

/** good_code.py */
import requests

# 環境変数にプロキシを設定するか、requestsに直接渡す
# import os
# os.environ['HTTP_PROXY'] = 'http://your.proxy.server:port'
# os.environ['HTTPS_PROXY'] = 'http://your.proxy.server:port'

proxies = {
    'http': 'http://your.proxy.server:port',
    'https': 'http://your.proxy.server:port'
} # 適切なプロキシ設定に置き換える

try:
    response = requests.get('https://api.external.com/items', proxies=proxies, timeout=10)
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
except requests.exceptions.Timeout as e:
    print(f"Timeout Error: {e}")

パターン3: TLS/SSL証明書の問題

/** bad_code.py */
import requests

try:
    # 自己署名証明書や期限切れの証明書を持つHTTPSサイトにアクセス
    response = requests.get('https://self-signed-cert.example.com/data') # 検証エラーが発生する可能性
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
    # 出力例:
    # Connection Error: HTTPSConnectionPool(...): Max retries exceeded with url: /data (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self signed certificate in certificate chain (_ssl.c:1129)')))

HTTPS接続において、対象サーバーのSSL/TLS証明書が無効、期限切れ、自己署名である場合、requestsはセキュリティ上の理由から接続を拒否し、SSLError を含む ConnectionError を発生させます。これは中間者攻撃を防ぐための重要な挙動です。

/** good_code.py */
import requests

try:
    # 信頼できる証明書を持つHTTPSサイトにアクセス
    response = requests.get('https://www.google.com', timeout=5)
    print(response.status_code)

    # もし自己署名証明書などを一時的に信頼したい場合（本番環境では非推奨）
    # response = requests.get('https://self-signed-cert.example.com/data', verify=False, timeout=5)
    # print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
except requests.exceptions.SSLError as e:
    print(f"SSL Error: {e}. Check certificate validity.")
except requests.exceptions.Timeout as e:
    print(f"Timeout Error: {e}")

よくあるバリエーション

requests.exceptions.ConnectionError: Max retries exceeded with url

このエラーメッセージは、requestsが指定されたURLへの接続を複数回試行したものの、すべて失敗したことを示します。根本的な原因としては、対象サーバーがダウンしている、URLが間違っている、またはネットワーク接続が完全にブロックされている場合がほとんどです。

/** コード例 */
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# リトライ回数を増やす設定（根本原因解決ではないが、一時的なネットワーク不安定に対応）
retry_strategy = Retry(
    total=5, # 最大リトライ回数
    backoff_factor=1, # 最初の失敗から1秒、2秒、4秒と待機
    status_forcelist=[429, 500, 502, 503, 504], # リトライするHTTPステータスコード
    allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
http = requests.Session()
http.mount("https://", adapter)
http.mount("http://", adapter)

try:
    # 存在しないURLへのアクセス
    response = http.get('http://non-existent-domain.com/data', timeout=3)
    print(response.status_code)
except requests.exceptions.ConnectionError as e:
    print(f"Caught ConnectionError after retries: {e}")

requests.exceptions.ConnectionError: RemoteDisconnected

RemoteDisconnected は、サーバー側が予期せず接続を閉じた場合に発生します。これは、サーバーがリクエストを処理しきれずにクラッシュした、またはアイドルタイムアウトで接続を切断した、あるいはファイアウォールやロードバランサーが接続を遮断した可能性を示唆します。

/** コード例 */
import requests

try:
    # サーバーが応答を返す前に接続を切断するような状況を想定
    response = requests.get('http://api.example.com/long-running-task', timeout=1)
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"Caught ConnectionError (RemoteDisconnected): {e}")
    # サーバーのログやインフラ設定（ロードバランサーのタイムアウトなど）を確認

requests.exceptions.ConnectionError: SSLError

このエラーは、HTTPS接続時にSSL/TLS証明書の検証に失敗した場合に発生します。証明書が無効、期限切れ、ホスト名と一致しない、または信頼されていない認証局によるものなどが原因です。セキュリティ上の理由から、デフォルトでrequestsはこのような接続を拒否します。

/** コード例 */
import requests

try:
    # 自己署名証明書を持つサイトへのアクセス
    response = requests.get('https://badssl.com/', timeout=5) # このURLはSSLエラーを意図的に引き起こす
    print(response.status_code)
except requests.exceptions.SSLError as e:
    print(f"Caught SSLError: {e}")
    # 必要に応じてSSL検証を無効にする (verify=False) ことも可能だが、本番では推奨されない
    # response = requests.get('https://badssl.com/', verify=False, timeout=5)
    # print(response.status_code)
except requests.exceptions.ConnectionError as e:
    print(f"Caught general ConnectionError: {e}")

フレームワーク別の発生パターン

Django (DRF)での発生パターン

Django REST Framework (DRF) を使ったバックエンドサービスが、決済ゲートウェイや外部認証サービスなどの外部APIと連携する際に requests.exceptions.ConnectionError が発生することがあります。例えば、ユーザー登録時に外部サービスに通知を送る処理や、定期的に外部からデータを取得するバッチ処理などで、ネットワークの問題やAPIのエンドポイントがダウンしている場合に遭遇します。

/** myapp/views.py (Django REST Framework) */
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
import requests
import os

EXTERNAL_API_URL = os.environ.get('EXTERNAL_API_URL', 'http://localhost:8001/external-service')

class ExternalServiceCallView(APIView):
    def post(self, request):
        payload = {"user_id": request.user.id, "data": request.data}
        try:
            # Bad: タイムアウト設定が不十分、またはURLが間違っている
            # response = requests.post(EXTERNAL_API_URL, json=payload)
            # Good: 適切なタイムアウトとエラーハンドリング
            response = requests.post(EXTERNAL_API_URL, json=payload, timeout=5)
            response.raise_for_status() # HTTPステータスコードがエラーの場合に例外を発生
            return Response(response.json(), status=status.HTTP_200_OK)
        except requests.exceptions.ConnectionError as e:
            print(f"Connection Error to external service: {e}")
            return Response({'error': '外部サービスに接続できませんでした。'}, status=status.HTTP_503_SERVICE_UNAVAILABLE)
        except requests.exceptions.Timeout as e:
            print(f"Timeout Error to external service: {e}")
            return Response({'error': '外部サービスの応答がタイムアウトしました。'}, status=status.HTTP_504_GATEWAY_TIMEOUT)
        except requests.exceptions.RequestException as e:
            print(f"Other Request Error to external service: {e}")
            return Response({'error': '外部サービスからのエラー応答。'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)

FastAPIでの発生パターン

FastAPIアプリケーションが起動時に外部設定（例: 設定サービスからの動的な値取得）をフェッチしたり、バックグラウンドタスクで外部APIを定期的に呼び出したりする際にこのエラーが発生することがあります。特に、コンテナ環境（Docker）でデプロイされている場合、コンテナ間のネットワーク設定やDNS解決の問題が原因で発生しやすいです。

/** main.py (FastAPI) */
from fastapi import FastAPI, HTTPException, BackgroundTasks
import requests
import os
import asyncio

app = FastAPI()

EXTERNAL_CONFIG_URL = os.environ.get('EXTERNAL_CONFIG_URL', 'http://localhost:8001/config')

@app.on_event("startup")
async def startup_event():
    print("Starting up...")
    try:
        # Bad: 外部設定サービスのURLが間違っているか、サービスが未起動
        # response = requests.get(EXTERNAL_CONFIG_URL, timeout=3)
        # Good: 適切なURLとエラーハンドリング
        response = requests.get(EXTERNAL_CONFIG_URL, timeout=3)
        response.raise_for_status()
        app.state.config = response.json()
        print(f"Loaded config: {app.state.config}")
    except requests.exceptions.ConnectionError as e:
        print(f"CRITICAL: Failed to connect to external config service: {e}")
        # 起動時に必須のサービスなら、アプリケーションの起動を中断することも検討
        # raise SystemExit(f"Cannot start without config: {e}")
    except requests.exceptions.Timeout as e:
        print(f"WARNING: Timeout connecting to external config service: {e}")
    except requests.exceptions.RequestException as e:
        print(f"ERROR: Failed to fetch external config: {e}")

@app.get("/config")
async def get_config():
    if hasattr(app.state, 'config'):
        return app.state.config
    raise HTTPException(status_code=503, detail="Config not loaded")

根本原因の特定方法

まず、エラーメッセージから具体的な原因（Max retries exceeded, RemoteDisconnected, SSLError など）を特定します。次に、pingコマンドで対象ドメインへの疎通確認、telnetコマンドで対象ポートへの接続確認を行います。もしプロキシ環境であれば、環境変数 HTTP_PROXY / HTTPS_PROXY が正しく設定されているか確認します。requestsライブラリのデバッグログを有効にすることで、詳細なリクエスト/レスポンス情報を取得できる場合もあります（例: import logging; logging.basicConfig(level=logging.DEBUG)）。また、対象サーバー側のアクセスログやエラーログも確認し、リクエストがサーバーに到達しているか、何かエラーが発生しているかを確認します。

/** Python requests のデバッグログを有効にする */
import logging
import http.client as http_client

# HTTPリクエストのデバッグログを有効化
http_client.HTTPConnection.debuglevel = 1

logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True

import requests

try:
    # 存在しないURLにアクセスして詳細なログを確認
    response = requests.get('http://non-existent-domain.com/test', timeout=5)
    print(response.status_code)
except requests.exceptions.ConnectionError as e:
    print(f"Connection Error: {e}")
except requests.exceptions.Timeout as e:
    print(f"Timeout Error: {e}")

/** シェルコマンドでの疎通確認 */
# ping コマンド
# ping api.example.com

# telnet コマンド（ポートへの接続確認）
# telnet api.example.com 443

タイムアウト設定の重要性とリトライ戦略

requestsライブラリで外部リソースにアクセスする際、タイムアウト設定は非常に重要です。デフォルトではタイムアウトが無期限であるため、サーバーが無応答の場合にプログラムがハングアップする可能性があります。timeout 引数を使用することで、接続確立と応答待ちの最大時間を設定できます。また、一時的なネットワークの不安定さやサーバー負荷によるエラーに対応するため、リトライ戦略を実装することも一般的です。requests の HTTPAdapter と urllib3.util.retry.Retry を組み合わせることで、自動リトライとバックオフ（再試行間隔の延長）を簡単に設定できます。

防止策とベストプラクティス

このエラーを防ぐためには、堅牢なネットワークリクエスト処理を実装することが重要です。具体的には、すべての requests 呼び出しに適切な timeout 値を設定し、try-exceptブロックで requests.exceptions.ConnectionError や requests.exceptions.Timeout を捕捉してエラーハンドリングを行います。一時的なネットワークの問題に対応するために、リトライロジック（requests の HTTPAdapter を利用）を導入することも有効です。また、APIのURLやプロキシ設定は環境変数で管理し、本番環境と開発環境で容易に切り替えられるようにすることで、設定ミスによるエラーを防止できます。

/** Python requests のリトライとタイムアウト設定 */
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import os

# 環境変数からAPI_URLを取得、なければデフォルト
API_URL = os.environ.get('EXTERNAL_API_URL', 'https://jsonplaceholder.typicode.com/posts/1')

# リトライ戦略の設定
retry_strategy = Retry(
    total=3, # 最大3回リトライ
    backoff_factor=1, # 1秒, 2秒, 4秒と待機
    status_forcelist=[429, 500, 502, 503, 504], # これらのステータスコードでリトライ
    allowed_methods=["GET", "POST"] # リトライ対象のHTTPメソッド
)
adapter = HTTPAdapter(max_retries=retry_strategy)

# requests.Session を使用してアダプターをマウント
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

try:
    # タイムアウトを5秒に設定
    response = session.get(API_URL, timeout=5)
    response.raise_for_status() # HTTPエラー（4xx, 5xx）があれば例外を発生させる
    print(response.json())
except requests.exceptions.ConnectionError as e:
    print(f"ネットワーク接続エラー: {e}")
    # ユーザーへのメッセージ表示やログ記録
except requests.exceptions.Timeout as e:
    print(f"リクエストがタイムアウトしました: {e}")
except requests.exceptions.RequestException as e:
    print(f"その他のリクエストエラー: {e}")
except Exception as e:
    print(f"予期せぬエラー: {e}")

よくある質問（FAQ）