api接口怎么对接?手把手带你走完对接全流程(附代码)

上一篇我们聊了 API 对接的整体思路和避坑经验,不少朋友反馈:道理都懂,真到动手的时候还是不知道第一行代码怎么写。这篇就用一个真实场景——把物流跟踪平台的数据拉进自己的 ERP——带着大家把对接的五个步骤完整走一遍,每一步都给出可运行的代码。
第一步:拿到并读懂 API 文档
文档是对接的基础,接口地址、请求方式、参数格式、返回结构、认证方式全在里面。拿到文档后别急着写代码,先做四件事:
1. 确认文档里有完整的示例请求和响应——只有参数表没有示例的文档,对接时一半时间在猜;
2. 检查版本号,确认用的是最新版接口说明,别对着过期文档调半天;
3. 重点看认证机制(API Key 还是 OAuth)和频率限制(比如每分钟最多 60 次),这两个直接决定代码怎么设计;
4. 把返回结构里的字段和自己的数据库字段先对照一遍,列出映射表。
第二步:配置身份认证
大多数 API 需要认证。最常见的是 API Key 方式,把密钥放进请求头即可:
headers = {
"Authorization": "Bearer sk_live_xxxxxxxx",
"Content-Type": "application/json"
}
如果对方用的是 OAuth 2.0,要先访问授权地址换取 access_token,token 通常有有效期(比如 2 小时),需要缓存并在过期前刷新:
def get_token():
r = requests.post(OAUTH_URL, data={
"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET,
"grant_type": "client_credentials"})
r.raise_for_status()
return r.json()["access_token"]
认证配置好之后,先调一个最简单的健康检查或查询类接口验证密钥是否生效,再往下走——不要把认证调试留到业务接口一起做,出问题分不清是密钥错了还是参数错了。
第三步:构造 HTTP 请求
按文档规则拼请求。以查询运单轨迹为例(GET 请求,运单号作为查询参数):
resp = requests.get(
"https://api.example.com/v1/track",
headers=headers,
params={"track_number": "SF1234567890"},
timeout=10
)
几个细节:请求方法要按文档来——GET 取数据、POST 提交数据,别混用;POST 请求的参数放请求体里,按 JSON 格式传(json= 参数);一定要设置 timeout,没有超时的请求在网络抖动时会永远挂着。
第四步:处理响应,把数据落库
拿到响应后,先判断状态码,再解析 JSON、提取字段写入自己的系统:
if resp.status_code == 200:
data = resp.json()
for node in data["trace_nodes"]:
save_to_db(
track_number=data["track_number"],
status=node["status"],
accept_time=node["accept_time"]
)
else:
log.error("API返回异常: %s %s", resp.status_code, resp.text)
注意:状态码 200 只代表请求成功,不代表业务成功——很多 API 在业务失败(比如运单号不存在)时也返回 200,真正的成败标志在响应体里的 code 字段,两层都要判断。
第五步:调试与日志记录
对接过程一定会出问题,调试效率取决于手段:
1. 写代码前,先用 Postman 或 curl 把接口跑通,确认参数格式没问题再翻译成代码,能省一半调试时间;
2. 代码里保留完整的请求和响应日志:时间戳、请求参数、响应状态码、错误消息,缺一不可;
3. 遇到超时先查网络连通性,遇到 401 先检查 token 是否过期,这两类问题占了对接故障的大头。
完整串起来:一个最小可运行的同步脚本
import requests, time
def sync_tracks(numbers):
session = requests.Session()
session.headers.update(auth_headers())
for num in numbers:
try:
r = session.get(API_URL, params={"track_number": num}, timeout=10)
if r.status_code == 200 and r.json()["code"] == 0:
save_to_db(r.json()["data"])
else:
log.error("sync fail: %s -> %s", num, r.text[:200])
except requests.RequestException as e:
log.error("network error: %s -> %s", num, e)
time.sleep(1) # 控制频率,避免触发限流
这几十行就是绝大多数对接项目的骨架:认证、请求、判断、落库、限频、日志全在里面。真实项目在此基础上再补充重试机制、增量同步和监控告警。
厦门纬度软件科技有限公司专注 API 接口开发与系统数据打通,RESTful、SOAP、GraphQL 三种协议均有成熟对接经验,支持私有化部署让数据不出内网。如果你不想自己折腾这些细节,欢迎联系我们,通常几天就能帮你把系统打通并交付带日志监控的完整方案。