给requests封装一个可复用的代理客户端，避免项目里到处写proxies

原创

能吃辣椒和砍树的熊大

发布于 2026-08-06 23:54:28

940

小型Python脚本中，直接在每个请求里写proxies问题不大。

项目逐渐变大以后，代码里经常会出现这样的内容：

requests.get(url_a, proxies=proxies)
requests.get(url_b, proxies=proxies)
requests.post(url_c, proxies=proxies)

代理配置、超时、请求头和异常处理散落在各个模块中，后续调整起来比较麻烦。

因此，我给项目封装了一个简单的代理客户端。

一、希望解决的问题

这个客户端需要统一处理：

代理地址；
HTTP和HTTPS配置；
连接超时；
请求超时；
Session连接复用；
重试；
请求头；
代理认证错误；
资源释放。

二、基础实现

from __future__ import annotations

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class ProxyHttpClient:
    def __init__(
        self,
        proxy_host: str,
        proxy_port: int,
        connect_timeout: int = 5,
        read_timeout: int = 20,
    ) -> None:
        self.timeout = (
            connect_timeout,
            read_timeout,
        )

        proxy_url = (
            f"http://{proxy_host}:{proxy_port}"
        )

        self.session = requests.Session()

        self.session.proxies.update({
            "http": proxy_url,
            "https": proxy_url,
        })

        self.session.headers.update({
            "User-Agent": "ProxyHttpClient/1.0",
        })

        retry = Retry(
            total=3,
            connect=3,
            read=2,
            status=2,
            backoff_factor=0.8,
            status_forcelist=(
                502,
                503,
                504,
            ),
            allowed_methods=(
                "GET",
                "HEAD",
                "OPTIONS",
            ),
        )

        adapter = HTTPAdapter(
            max_retries=retry,
            pool_connections=20,
            pool_maxsize=20,
        )

        self.session.mount(
            "http://",
            adapter,
        )
        self.session.mount(
            "https://",
            adapter,
        )

三、统一封装GET和POST

    def get(
        self,
        url: str,
        **kwargs,
    ) -> requests.Response:
        kwargs.setdefault(
            "timeout",
            self.timeout,
        )

        response = self.session.get(
            url,
            **kwargs,
        )

        self._check_response(response)
        return response

    def post(
        self,
        url: str,
        **kwargs,
    ) -> requests.Response:
        kwargs.setdefault(
            "timeout",
            self.timeout,
        )

        response = self.session.post(
            url,
            **kwargs,
        )

        self._check_response(response)
        return response

统一状态检查：

    @staticmethod
    def _check_response(
        response: requests.Response,
    ) -> None:
        if response.status_code == 407:
            raise RuntimeError(
                "代理认证失败"
            )

        response.raise_for_status()

最后增加关闭方法：

    def close(self) -> None:
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(
        self,
        exc_type,
        exc_value,
        traceback,
    ):
        self.close()

四、实际使用

with ProxyHttpClient(
    proxy_host="127.0.0.1",
    proxy_port=8080,
) as client:
    response = client.get(
        "https://httpbin.org/ip"
    )

    print(response.text)

业务代码中不再需要关心：

代理字典怎么写；
Session如何创建；
连接池怎么配置；
何时关闭连接。

五、为什么只重试部分状态码？

这里没有把所有非200状态都加入重试。

例如：

400 Bad Request
401 Unauthorized
404 Not Found

这些通常是请求参数、权限或URL错误，重复发送并不能解决问题。

而：

502
503
504

更可能属于临时网络或服务异常，可以在有限次数内重试。

六、POST请求为什么默认不重试？

POST请求可能会产生实际业务操作。

如果第一次请求已经在服务端执行，但客户端没有收到响应，再次自动重试可能导致重复提交。

因此，示例中的自动重试默认只针对：

GET、HEAD、OPTIONS

需要重试POST时，应为业务增加幂等键或者去重逻辑。

七、固定代理与短效代理的区别

这个客户端更适合：

长效代理；
固定IP；
固定入口隧道代理。

如果使用短效代理API，需要增加代理刷新能力。可以把代理获取部分抽象成一个策略接口，而不是把代理地址写死在构造函数中。

总结

封装代理客户端的价值不是减少几行代码，而是把网络请求相关配置集中管理。

后续需要调整：

超时时间；
重试次数；
请求头；
认证；
日志；
连接池；

都可以在一个位置完成。

本文接入思路可以用于携趣长效代理或固定入口的隧道代理。具体IP、端口和授权信息应从产品后台获取，不建议把真实参数直接写进代码仓库。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

tcp/ip