RESTful API 集成：接口调用规范与数据流转

说实话，RESTful API 这个词，你可能早就听烂了。但真正在项目里对接过十几个第三方接口之后，你会发现——懂概念和能写好代码，中间差着好几个坑。我个人习惯把 REST 调用拆成三块看：怎么发请求、怎么传参数、怎么接响应。每块都有一些“不说就会翻车”的细节，咱们一个一个捋清楚。

RESTful API 调用核心流程 构造请求 URL + Method + Headers 设置参数 Query / Body / Path 发起请求 requests / httpx 状态码判断 2xx / 4xx / 5xx 解析响应体 JSON / 错误码提取 业务处理 数据映射 / 重试 ⚠️ 注意：所有网络 I/O 都必须配置超时与重试机制，否则生产环境分分钟教你做人。 我曾经就因为忘了设超时，一个第三方接口挂了，拖垮了整个支付线。

1. 请求构造：别只拼 URL

写一个 requests.get() 很容易，但要写好一套健壮的调用，你得先想清楚三个东西：

  • HTTP 方法：GET 查、POST 增、PUT 全量更新、PATCH 局部更新、DELETE 删。别在 POST 接口里传 id 做删除，这很危险。
  • 请求头：Content-Type 必须明确，常见的有 application/json 或 multipart/form-data。还有 Authorization，Bearer 放在 header 里，别傻傻拼在 URL 上。
  • 超时设置：这一条我换过好几次跟头。不加 timeout，网络抖动时你的线程可能挂几分钟。建议 connect 5s，read 15s。
import requests

url = "https://api.example.com/v2/users"
headers = {
    "Authorization": "Bearer your__here",
    "Content-Type": "application/json"
}
# 超时设置：连接 3 秒，读取 10 秒
resp = requests.get(url, headers=headers, timeout=(3, 10))
个人习惯：我会把所有请求配置封装成一个 Session 对象，统一设置 headers 和 retry 策略，避免每个调用重复写 。

2. 参数传递：三个门路，别走错

参数怎么传，取决于接口设计者的约定。常见的有三种：

参数位置适用场景Python 写法
Query（查询参数）GET 请求，分页、筛选、排序params={"page":1, "size":20}
Path（路径参数）资源标识，如 /users/{id}f-string 或 url.format(id=123)
Body（请求体）POST/PUT/PATCH 提交数据json={"name":"张三"}

你想想看，如果接口要求是 JSON 格式，你却传了 form-data，可能收到 415 或诡异的错误。我以前对接一个物流接口，调试了半天，最后发现是 Content-Type 写错了。

# Query 参数示例
params = {"page": 1, "page_size": 50, "sort": "-created_at"}
resp = requests.get(url, headers=headers, params=params, timeout=5)

# Body JSON 示例
payload = {"username": "alice", "email": "alice@example.com"}
resp = requests.post(url, headers=headers, json=payload, timeout=5)
避坑指南：我曾经在一个项目里用 data=payload 传 JSON，结果 requests 库自动把 dict 转成了 form-encoded，对方服务器识别不了。之后我强制用 json=payload，再也没踩过这个坑。

3. 响应处理：别只读 .json()

拿到 response 对象，第一件事不是 .json()，而是检查状态码。我见过新手直接 .json() 然后报错，因为服务器返回了 502 页面。

resp = requests.get(url, headers=headers, timeout=5)
if resp.status_code == 200:
    data = resp.json()
    # 正常处理
elif resp.status_code == 401:
    #  过期，需要刷新
    refresh_()
elif resp.status_code == 429:
    # 被限流，等待后重试
    time.sleep(5)
    retry()
else:
    # 记录错误日志
    print(f"Unexpected status {resp.status_code}: {resp.text}")

另外，响应的结构通常有固定 schema。比如 SWAGGER 规范的返回值：

{
  "code": 0,
  "message": "success",
  "data": { ... }
}
// 或分页格式
{
  "items": [...],
  "total": 100,
  "page": 1
}

我建议你写一个通用的响应解析函数，统一提取 code 和 data，而不是在每个调用里手写 if-else。代码复用率高，也方便统一处理业务错误。

def parse_response(resp: requests.Response) -> dict:
    """通用响应解析，失败时抛出异常"""
    if resp.status_code != 200:
        raise ApiException(f"HTTP {resp.status_code}: {resp.text}")
    body = resp.json()
    if body.get("code") != 0:
        raise ApiException(f"业务错误 {body.get('code')}: {body.get('message')}")
    return body.get("data")
核心原则：永远假设网络会失败、接口会返回异常数据。防御性编程在 API 集成中不可或缺。

4. 重试与容错：生产环境必修课

我不可能保证每次请求都成功。网络波动、服务重启、限流——这些都太常见了。Python 的 requests 库自带的 HTTPAdapter 可以配置重试策略，或者直接用 tenacity 库更精细。

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

session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"]  # 幂等方法才重试
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)

# 之后用 session 发起请求，自动重试
resp = session.get(url, headers=headers, timeout=5)
注意：POST 请求要小心重试。如果接口不是幂等的（比如创建订单），重试可能导致重复下单。解决办法是客户端生成唯一 idempotency-key 并传给服务端。

5. 异步调用：当性能不够用

如果一次请求耗时 100ms，串行调 10 个外部接口就是 1 秒。你的用户可等不起。Python 的 httpx 或 aiohttp 可以发起并发请求，显著提升吞吐。

import httpx
import asyncio

async def fetch_user(client, user_id):
    url = f"https://api.example.com/users/{user_id}"
    resp = await client.get(url, timeout=10)
    return resp.json()

async def main():
    async with httpx.AsyncClient() as client:
        tasks = [fetch_user(client, uid) for uid in [1,2,3,4,5]]
        results = await asyncio.gather(*tasks)
        for user in results:
            print(user["name"])

# asyncio.run(main())

嗯，这里要注意：异步代码里同样要处理重试和超时，而且有的库不支持 async 下的自动重试，你需要自己 wrap。

重要提醒：不要无限制并发。给 asyncio.Semaphore 设个上限，比如 10，否则你可能被对方服务器封 IP。

6. 一个完整的调用示例

说了这么多，来个真实的综合案例吧：查询用户列表，带分页，处理错误，重试一次。

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

def build_session(: str) -> requests.Session:
    session = requests.Session()
    session.headers.update({
        "Authorization": f"Bearer {}",
        "Content-Type": "application/json"
    })
    retries = Retry(total=2, backoff_factor=0.5, status_forcelist=[429, 500, 502])
    session.mount("https://", HTTPAdapter(max_retries=retries))
    return session

def get_users(session, page=1, size=20):
    url = "https://api.example.com/v2/users"
    params = {"page": page, "size": size}
    try:
        resp = session.get(url, params=params, timeout=(3, 10))
        resp.raise_for_status()
        data = resp.json()
        # 假设返回 { "code":0, "data":{"items":[], "total":100} }
        if data.get("code") != 0:
            raise ValueError(f"业务错误: {data.get('message')}")
        return data["data"]
    except requests.exceptions.RequestException as e:
        print(f"请求异常: {e}")
        return None

# 使用
 = "your_"
session = build_session()
users = get_users(session, page=1)
if users:
    print(f"获取到 {len(users['items'])} 个用户，共 {users['total']} 个")

说实话，这些代码看起来不复杂，但每一条都是我在生产环境里摔出来的经验。RESTful API 集成，本质上就是管好三个事：怎么发、发什么、怎么接。再加上容错和异步，基本就能应付大多数场景了。

← 上一章 📖 返回目录 下一章 →