## 为什么写这篇

团队里对接的第三方服务越来越多:短信、推送、支付、地图、OCR……每接一个,代码里就多一坨 `requests.post(...)`。超时没设、异常裸奔、签名逻辑复制粘贴,出了问题全靠 grep 日志。上个月排查一个"短信偶发发不出去"的问题,翻了半天才发现是某个角落里的调用没设超时,worker 被拖死后消息静默丢了。

实在受不了了,把这几年对接第三方 API 的封装套路整理了一遍,沉淀成一个基类。现在新接一个渠道,写个子类描述清楚接口参数,半小时就能上线。这篇把整套模式分享出来,代码都是生产环境在跑的版本(脱敏过)。

## 裸奔的代码长什么样

先看不封装时的典型写法,应该很眼熟:

```python
import requests

def send_sms(phone, content):
    resp = requests.post(
        "https://api.example.com/v1/sms/send",
        json={"phone": phone, "content": content, "app_key": "ak_123"},
    )
    data = resp.json()
    if data["code"] == 0:
        return data["data"]["msg_id"]
    return None
```

问题一抓一大把:

- **没设超时**。对方服务挂了你这边的 worker 就跟着吊死,还是挂一片的那种
- **网络抖动直接抛 ConnectionError**,调用方根本分不清"要不要重试"
- **业务错误码被静默吞掉**,返回 `None`,上游不知道是没发送还是发送失败
- **密钥硬编码**,换个环境改代码,迟早把生产密钥提交进仓库
- 如果对方要签名,签名逻辑散在每个函数里,改签名算法要改七八处

同一个套路复制五六份之后,任何一次统一改造(比如加灰度 header、换网关)都是体力活。

## 封装思路:三层切分

不绕弯子,直接说结构:

1. **配置层**:`base_url`、密钥、超时、重试次数,全部收进一个配置对象,从环境变量读
2. **客户端层(基类)**:管 Session、连接池、签名、重试、异常转换、日志,不碰具体业务
3. **业务层(子类)**:每个第三方平台一个子类,只描述"这个平台有哪些接口、参数长什么样"

调用方最后只面对三件事:调用方法、处理三种异常。下面逐块给代码,最后有完整版。

## 第一步:配置和异常体系

配置用 `dataclass` 就够了,别一上来就引入配置框架:

```python
from dataclasses import dataclass

@dataclass
class APIConfig:
    base_url: str                    # 接口地址,如 https://api.example.com
    app_key: str = ""
    app_secret: str = ""
    connect_timeout: float = 3.0     # 连接超时:对方网络有问题时快速失败
    read_timeout: float = 10.0       # 读取超时:对方处理慢时别让 worker 干等
    max_retries: int = 2             # 网络层失败的最大重试次数(不含首次)
    retry_backoff: float = 0.5       # 退避基数(秒),第 n 次重试前等 backoff * 2^n
    verify_ssl: bool = True
```

异常体系是整个封装的灵魂。核心诉求是:**让调用方能根据异常类型做决定,而不是去解析字符串**。三种就够:

```python
class APIError(Exception):
    """所有第三方调用异常的基类,上层只 catch 这一个也够用"""


class APINetworkError(APIError):
    """网络层失败:连不上、超时、DNS 解析失败等。已经重试过且耗尽了,可以走兜底"""


class APIResponseError(APIError):
    """对方给了明确答复,但结果不符合预期(HTTP 非 200 / 业务码非成功)"""

    def __init__(self, message, *, status_code=0, biz_code=None, payload=None):
        super().__init__(message)
        self.status_code = status_code   # HTTP 状态码
        self.biz_code = biz_code         # 对方业务错误码
        self.payload = payload           # 原始响应,排查问题时救命用


class APIAuthError(APIResponseError):
    """鉴权失败。一般是密钥配错了,重试一百次也没用,单独拎出来"""
```

`payload` 字段是我踩坑后加的。以前只记 message,对方返回格式一变就两眼一抹黑;现在把原始响应留着,排查效率完全不是一个量级。

## 第二步:基类,把脏活全收进来

核心是一个 `BaseAPIClient`。几个设计决定先说清楚:

- **重试自己写循环**,不用 urllib3 的 `Retry`。原因:`Retry` 管不到业务错误码,也没法区分"这个 POST 不是幂等的,不能重试"。自己写几十行,控制力完全在自己手里
- **默认只对幂等方法重试**(GET/PUT/DELETE),POST 创建类操作默认不重试,除非调用方显式声明幂等
- **业务码错误不重试**。对方已经明确告诉你"余额不足"了,重试只会再打一遍同样的脸

```python
import hashlib
import hmac
import json
import logging
import time
import uuid
from typing import Any, Optional

import requests
from requests import Response, Session
from requests.adapters import HTTPAdapter


class BaseAPIClient:
    # 各家平台"业务成功"的标志不一样,按需覆盖
    SUCCESS_CODES = {0, "0", "success", "SUCCESS", "ok", "OK"}
    # 业务码里的鉴权失败,按对接文档维护
    AUTH_BIZ_CODES = {"SIGN_ERROR", "INVALID_TOKEN", "10001", 10001}

    def __init__(self, config: APIConfig, logger: Optional[logging.Logger] = None):
        self.config = config
        self.log = logger or logging.getLogger(self.__class__.__name__)
        self.session = self._build_session()

    def _build_session(self) -> Session:
        s = requests.Session()
        # 默认连接池只有 10,并发高的场景务必调大
        adapter = HTTPAdapter(pool_connections=10, pool_maxsize=20)
        s.mount("http://", adapter)
        s.mount("https://", adapter)
        return s
```

签名这块,各家规则千差万别(排序方式、拼接符、是否包含 body、HMAC 还是 RSA),但骨架都差不多。给一个最常见的 HMAC-SHA256 版本,子类按需覆盖:

```python
    def _sign(self, params: dict, body: Optional[dict],
              timestamp: str, nonce: str) -> str:
        raw = {
            "app_key": self.config.app_key,
            "timestamp": timestamp,
            "nonce": nonce,
            **(params or {}),
            **(body or {}),
        }
        # 按 key 排序拼成 k1=v1&k2=v2,再用 secret 做 HMAC
        canonical = "&".join(f"{k}={raw[k]}" for k in sorted(raw))
        return hmac.new(
            self.config.app_secret.encode(),
            canonical.encode(),
            hashlib.sha256,
        ).hexdigest()

    def _auth_headers(self, params: dict, body: Optional[dict]) -> dict:
        ts = str(int(time.time()))
        nonce = uuid.uuid4().hex[:16]
        return {
            "X-App-Key": self.config.app_key,
            "X-Timestamp": ts,
            "X-Nonce": nonce,
            "X-Sign": self._sign(params, body, ts, nonce),
        }
```

然后是整个基类的心脏,统一的请求入口:

```python
    def _request(self, method: str, path: str, *,
                 params: dict = None, body: dict = None,
                 need_auth: bool = True,
                 idempotent: bool = None) -> Any:
        url = self.config.base_url.rstrip("/") + "/" + path.lstrip("/")
        headers = self._auth_headers(params, body) if need_auth else {}

        # 默认幂等方法才可重试;调用方可显式覆盖
        if idempotent is None:
            idempotent = method.upper() in ("GET", "HEAD", "PUT", "DELETE")
        attempts = self.config.max_retries + 1 if idempotent else 1

        last_err: Optional[Exception] = None
        for i in range(attempts):
            try:
                resp = self.session.request(
                    method, url,
                    params=params,
                    json=body,
                    headers=headers,
                    timeout=(self.config.connect_timeout,
                             self.config.read_timeout),
                    verify=self.config.verify_ssl,
                )
                return self._parse(resp)
            except APIResponseError:
                # 对方已给出明确答复,重试结果一样,直接抛
                raise
            except requests.RequestException as e:
                # 网络层错误:按指数退避重试
                last_err = e
                if i < attempts - 1:
                    wait = self.config.retry_backoff * (2 ** i)
                    self.log.warning(
                        "request failed, retry %d/%d after %.1fs: %s %s, err=%s",
                        i + 1, attempts - 1, wait, method, url, e)
                    time.sleep(wait)

        raise APINetworkError(
            f"request to {url} failed after {attempts} attempts") from last_err
```

统一的响应解析。绝大多数平台都是 `{"code":..., "message":..., "data":...}` 结构,基类按这个写;遇到返回 XML 或者结构奇葩的,子类覆盖 `_parse` 一个方法就行:

```python
    def _parse(self, resp: Response) -> Any:
        if resp.status_code in (401, 403):
            raise APIAuthError("auth failed",
                               status_code=resp.status_code,
                               payload=resp.text[:500])
        if resp.status_code != 200:
            raise APIResponseError(
                f"unexpected http status: {resp.status_code}",
                status_code=resp.status_code, payload=resp.text[:500])

        try:
            data = resp.json()
        except json.JSONDecodeError:
            raise APIResponseError("response is not valid json",
                                   status_code=resp.status_code,
                                   payload=resp.text[:500])

        code = data.get("code")
        if code in self.SUCCESS_CODES:
            return data.get("data")
        if code in self.AUTH_BIZ_CODES:
            raise APIAuthError(str(data.get("message")),
                               biz_code=code, payload=data)
        raise APIResponseError(str(data.get("message")),
                               status_code=200, biz_code=code, payload=data)
```

最后补上便捷方法和资源释放:

```python
    def get(self, path, params=None, **kw):
        return self._request("GET", path, params=params, **kw)

    def post(self, path, body=None, idempotent=False, **kw):
        return self._request("POST", path, body=body,
                             idempotent=idempotent, **kw)

    def close(self):
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()
```

## 第三步:业务子类,只剩"描述接口"

基类就位之后,接一个新平台就只剩一件事:把对方的接口文档翻译成一个类。

```python
import os

class SmsClient(BaseAPIClient):
    """某短信平台(示例)"""

    def send(self, phone: str, template_id: str, params: dict) -> str:
        # 发送不是幂等操作,默认不重试;对方支持 msg_no 去重的话可以传 idempotent=True
        data = self.post("/v1/messages", body={
            "phone": phone,
            "template_id": template_id,
            "params": params,
        })
        return data["msg_id"]

    def query_status(self, msg_id: str) -> str:
        data = self.get(f"/v1/messages/{msg_id}")
        return data["status"]
```

调用方长这样:

```python
config = APIConfig(
    base_url="https://api.example-sms.com",
    app_key=os.environ["SMS_APP_KEY"],
    app_secret=os.environ["SMS_APP_SECRET"],
    read_timeout=5,
)

with SmsClient(config) as sms:
    try:
        msg_id = sms.send("138****1234", "LOGIN_CODE", {"code": "123456"})
    except APIAuthError:
        ...  # 密钥/签名问题,别重试,直接告警查配置
    except APINetworkError:
        ...  # 重试耗尽,走兜底:切备用渠道或入队稍后补发
    except APIResponseError as e:
        ...  # 对方明确拒绝,按 e.biz_code 对照错误码表做业务处理
```

对比一下开头那段裸奔代码:调用方只面对三种异常,每种的处理策略都是明确的。这就是这套封装值回票价的地方。

## 测试怎么写

不引入额外的 mock 库,`unittest.mock` 直接糊 `session.request` 就行:

```python
import json
import requests
from unittest import TestCase, mock


class SmsClientTest(TestCase):

    def setUp(self):
        self.client = SmsClient(APIConfig(
            base_url="https://api.test", app_key="k", app_secret="s"))

    def _resp(self, payload, status=200):
        r = mock.Mock()
        r.status_code = status
        r.text = json.dumps(payload)
        r.json.return_value = payload
        return r

    def test_send_success(self):
        ok = self._resp({"code": 0, "message": "ok", "data": {"msg_id": "m1"}})
        with mock.patch.object(self.client.session, "request", return_value=ok):
            self.assertEqual(
                self.client.send("13800001234", "T1", {}), "m1")

    def test_biz_error_no_retry(self):
        # 业务错误:不重试,直接抛
        err = self._resp({"code": 20001, "message": "balance not enough"})
        with mock.patch.object(self.client.session, "request",
                               return_value=err) as m:
            with self.assertRaises(APIResponseError) as ctx:
                self.client.query_status("m1")
            self.assertEqual(ctx.exception.biz_code, 20001)
            self.assertEqual(m.call_count, 1)

    def test_retry_on_timeout_then_success(self):
        # 前两次超时,第三次成功:GET 幂等,应当重试
        ok = self._resp({"code": 0, "message": "ok", "data": {"status": "done"}})
        side = [requests.ConnectTimeout(), requests.ConnectTimeout(), ok]
        with mock.patch.object(self.client.session, "request",
                               side_effect=side), \
             mock.patch("time.sleep"):  # 测试里别真睡
            self.assertEqual(self.client.query_status("m1"), "done")

    def test_post_not_retried(self):
        # POST 默认不幂等:超时应直接失败,不重复发
        with mock.patch.object(self.client.session, "request",
                               side_effect=requests.ConnectTimeout()) as m:
            with self.assertRaises(APINetworkError):
                self.client.send("13800001234", "T1", {})
            self.assertEqual(m.call_count, 1)
```

重点测四个行为:成功路径、业务错误不重试、幂等接口网络错误会重试、非幂等接口不重试。这四个守住了,基类的行为就稳了。

## 一些零碎但重要的经验

最后几条都是拿钱买来的教训:

**1. 日志必须脱敏。** 手机号中间四位打码,`app_secret` 和签名原文永远不进日志。出过一次事故,日志平台里能搜到完整手机号,整改了一个星期。

**2. 连接超时和读超时分开设,别一刀切。** 连接超时 3 秒足够,连不上就是连不上;读超时按接口的 SLA 来,查询类给短点,批量类给长点。

**3. POST 重试的前提是对方支持幂等。** 要么对方接口本身幂等(按业务单号去重),要么你自己生成幂等键传过去。两条都不满足,宁可失败报警人工介入,也不要赌"对方应该没收到"。重复扣款、重复发短信,哪个都够喝一壶。

**4. 维护一张错误码映射表。** 对接文档里的错误码,挑出现频率高的(余额不足、频率超限、参数错误)单独映射成自定义异常或枚举,剩下的走通用 `APIResponseError`。别试图全覆盖,几百个错误码你维护不动。

**5. 别开 `verify=False`。** 内网自签名证书就把 `verify` 指向你们的 CA 文件,关掉校验等于把 HTTPS 的裤子脱了。requests 对关闭校验的每次调用都会打 warning,看到那行字就该心虚。

**6. 高并发场景记得调连接池。** `Session` 默认 `pool_maxsize=10`,worker 并发一高,连接建了又拆,TIME_WAIT 堆成山。按实际并发量调 `HTTPAdapter`。

## 结尾

这套模式没有引入任何框架,核心就是一个基类加三种异常,总共两百来行。它的价值不在代码本身,而在于把"第三方调用"这件事的不确定性收敛到了一个地方:网络会抖、对方会挂、错误码千奇百怪,但你的业务代码面对的世界是干净的。

新接一个渠道,写个子类;对方改签名算法,改一个方法;要加统一的灰度 header,改一个地方。这大概就是"封装"两个字最朴素的意义。

团队里对接的第三方服务越来越多:短信、推送、支付、地图、OCR……每接一个,代码里就多一坨 requests.post(...)。超时没设、异常裸奔、签名逻辑复制粘贴,出了问题全靠 grep 日志。上个月排查一个"短信偶发发不出去"的问题,翻了半天才发现是某个角落里的调用没设超时,worker 被拖死后消息静默丢了。

第三方API对接的通用封装模式

编程语言

后端

开发工具

本文分享了一套高效封装第三方API调用的实战方案，通过三层架构设计（配置层、客户端层、业务层）解决超时缺失、异常裸奔、签名逻辑重复等问题。提供完整的异常体系、自动重试机制和日志脱敏方案，使新渠道接入时间缩短至半小时，显著提升系统稳定性和开发效率。

代码解释

短信

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

第三方API对接的通用封装模式-腾讯云开发者社区-腾讯云

第三方API对接的通用封装模式

第三方API对接的通用封装模式

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐