You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何解决Python中HTTPSConnectionPool的Max retries exceeded错误?

问题描述

requests代码本地运行正常,但部署到服务器后出现连接失败错误:

ConnectionError at /api/check-car/
HTTPSConnectionPool(host='example.uz', port=443): Max retries exceeded with url: /api/token/ (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7f3d48892f50>: Failed to establish a new connection: [Errno 111] Connection refused'))

涉及的核心代码如下:

def post(self, request):
    base_url = os.environ.get('AUTH_URL', 'http://localhost')
    # 向认证端点发送POST请求

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                      "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
        'accept': 'application/json',
        'Content-Type': 'application/json',
    }

    json_data = {
        'username': os.environ.get("GARAJ_USERNAME", "admin"),
        'password': os.environ.get("GARAJ_PASSWORD", "admin123"),
    }
    response = requests.post(base_url + '/api/token/', json=json_data, headers=headers)

    # 检查响应状态码
    if response.status_code == 200:
        data = response.json()
        # 从响应中提取JWT token
        jwt_token_access = data['access']
        # 在后续请求中携带JWT token
        headers['Authorization'] = f'Bearer {jwt_token_access}'
        auth_url = base_url + '/api/boshqarma/avto/'
        response = requests.get(auth_url, headers=headers)

        for x in response.json()['results']:
            if x["davlat_raqami"] == request.data['avto_raqami']:
                boshqarma = x["boshqarma"]
                response = requests.get(base_url + f'/api/boshqarma/boshqarma/{boshqarma}/', headers=headers)
                tuman = response.json()['tuman']
                response = requests.get(base_url + f'/api/boshqarma/bolim/{tuman}/', headers=headers)
                tuman = response.json()['tuman']
                tuman = Tuman.objects.get(tuman_nomi=tuman)
                data = {'tuman_id': tuman.id}
                response = Response(data)
                return response
        else:
            return Response("Xato avto raqam")

    else:
        return Response(response)
解决方案
  • 排查服务器间网络连通性:在部署代码的客户端服务器上执行curl -v https://example.uz/api/token/,确认能否和宿主服务器建立连接。如果连接被拒绝,重点检查:
    • 宿主服务器的防火墙/安全组是否允许客户端服务器的IP访问443端口;
    • 宿主服务器的Web服务(如Nginx、Apache)是否正常监听443端口,服务是否运行;
    • 宿主服务器的HTTPS证书是否有效,避免因证书验证失败导致连接中断。
  • 验证环境变量配置:确认服务器上的AUTH_URL是否正确设置为https://example.uz(而非本地默认的http://localhost),同时检查GARAJ_USERNAME和GARAJ_PASSWORD是否与宿主服务器的认证凭据匹配。
  • 添加超时与重试机制:原代码未设置请求超时,且无重试策略,容易因临时网络问题触发错误。可以修改代码使用Session配置重试,并添加超时参数:
    import requests
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry
    
    def post(self, request):
        base_url = os.environ.get('AUTH_URL', 'http://localhost')
        # 配置Session和重试策略
        session = requests.Session()
        retry_strategy = Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        session.mount("http://", adapter)
    
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            'accept': 'application/json',
            'Content-Type': 'application/json',
        }
    
        json_data = {
            'username': os.environ.get("GARAJ_USERNAME", "admin"),
            'password': os.environ.get("GARAJ_PASSWORD", "admin123"),
        }
        # 添加超时参数,避免请求挂起
        token_response = session.post(base_url + '/api/token/', json=json_data, headers=headers, timeout=10)
    
        if token_response.status_code == 200:
            data = token_response.json()
            jwt_token_access = data['access']
            headers['Authorization'] = f'Bearer {jwt_token_access}'
            auth_url = base_url + '/api/boshqarma/avto/'
            # 区分不同请求的响应变量,避免覆盖
            avto_response = session.get(auth_url, headers=headers, timeout=10)
    
            for x in avto_response.json()['results']:
                if x["davlat_raqami"] == request.data['avto_raqami']:
                    boshqarma = x["boshqarma"]
                    boshqarma_response = session.get(base_url + f'/api/boshqarma/boshqarma/{boshqarma}/', headers=headers, timeout=10)
                    tuman = boshqarma_response.json()['tuman']
                    bolim_response = session.get(base_url + f'/api/boshqarma/bolim/{tuman}/', headers=headers, timeout=10)
                    tuman = bolim_response.json()['tuman']
                    tuman = Tuman.objects.get(tuman_nomi=tuman)
                    return Response({'tuman_id': tuman.id})
            else:
                return Response("Xato avto raqam")
    
        else:
            return Response(token_response.text, status=token_response.status_code)
    
  • 检查宿主服务器服务状态:登录宿主服务器,查看提供/api/token/的认证服务是否正常运行,通过systemctl status <服务名称>查看状态,或查看服务日志定位内部错误。
  • 修复变量覆盖问题:原代码多次用response变量存储不同请求结果,容易导致调试困难。建议给每个请求的响应变量单独命名(如token_response、avto_response),提升代码可读性和可维护性。

内容的提问来源于stack exchange,提问作者Jakhongir Turgunboev

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.18 14:44:56