从 curl 到封装：腾讯天气 API 的工程化接入指南
为什么需要从 curl 走向工程封装
在调试阶段，一条简单的curl命令就能验证接口是否通畅、返回数据是否合理。但一旦要将天气预报、生活指数等功能集成到生产系统里，curl就远远不够了——你需要处理网络抖动时的重试、接口限流、参数校验、日志记录、响应解析异常等。本文以腾讯天气 API 为例，展示如何从原型级的curl一步一步过渡到一个可维护、可扩展的工程封装。
接口能力与适用场景
腾讯天气 API 提供了基于中文省市名的天气数据查询，无需经纬度坐标。主要能力包括：
- 实时天气：温度、湿度、风向风力、天气现象、更新时间
- 空气质量：AQI、PM2.5、PM10、质量等级
- 未来 7 天预报：每日最高/最低温度
- 24 小时逐时预报：每个整点的温度和天气
- 23 项生活指数：穿衣、紫外线、洗车、运动等
- 日出日落时间：每日的具体时刻
- 机动车限行：根据城市及区县返回限行尾号
典型使用场景：
- 智能家居控制面板显示室外天气
- 旅游 App 提供目的地未来一周天气概览
- 物流调度系统结合天气与限行规划路线
- 个人助手自动推送当日穿衣建议和限行提醒
请求参数与鉴权
接口基本信息
| 项目 | 内容 |
|---|---|
| 请求方法 | POST |
| 请求地址 | https://v1.apizero.cn/api/tencent-weather |
| 数据格式 | JSON |
| QPS 上限 | 10 次/秒 |
Header 参数
| 参数名 | 必填 | 类型 | 说明 |
|---|---|---|---|
| Authorization | 否 | string | Bearer <你的 API Key>，不传时使用默认匿名额度（较低） |
注意：官方文档中也可使用
X-API-Key头部传递密钥，两种方式等价，选择其一即可。
Body 参数（JSON）
| 字段名 | 必填 | 类型 | 描述 | 示例值 |
|---|---|---|---|---|
| province | 是 | string | 省 / 直辖市中文名 | 广东 |
| city | 是 | string | 市中文名 | 深圳 |
| county | 否 | string | 区 / 县中文名，提升定位精度及限行准确度 | 南山 |
{ "province": "广东", "city": "深圳", "county": "南山" }curl 快速验证
curl -sS \ -X POST \ -H "X-API-Key: $APIZERO_API_KEY" \ -H "Content-Type: application/json" \ -d '{"province": "广东", "city": "深圳", "county": "南山"}' \ "https://v1.apizero.cn/api/tencent-weather"如果返回的 JSON 中code为 0，则表示请求成功。若未传入 API Key，匿名额度为每日 500 次，通常也足够调试。
响应字段解读
成功响应示例（已精简）：
{ "code": 0, "msg": "成功", "request_id": "abc123", "data": { "observe": { "temperature": 30, "weather": "多云", "humidity": 77, "wind_direction": "北风", "wind_power": "3-4", "update_time": "2026-07-01 10:15" }, "air": { "aqi": 13, "level": 1, "quality": "优", "pm25": 4, "pm10": 12 }, "daily_forecast": [ { "date": "2026-07-01", "temperature": { "max": 33, "min": 26 } } ], "hourly_forecast": [ { "time": "07-01 10:00", "temperature": 30, "weather": "多云" } ], "life_index": [ { "key": "clothes", "name": "穿衣", "level": "炎热", "detail": "建议穿着轻薄衣物" } ], "sunrise_sunset": [ { "date": "2026-07-01", "sunrise": "05:43", "sunset": "19:12" } ], "limit": { "date": "2026-07-01", "tail_number": "3和8" }, "location": { "province": "广东", "city": "深圳", "county": "南山" }, "alarm": [] } }关键字段说明
| 顶层字段 | 说明 |
|---|---|
| code | 状态码，0 表示成功，非 0 表示异常 |
| msg | 状态描述 |
| request_id | 请求唯一标识，可用于排查问题 |
| data.observe | 实时观测数据 |
| data.air | 空气质量 |
| data.daily_forecast | 未来 7 天预报，数组 |
| data.hourly_forecast | 未来 24 小时逐时预报，数组 |
| data.life_index | 生活指数，数组 |
| data.sunrise_sunset | 日出日落时间，数组 |
| data.limit | 限行信息，若无限制可能返回空对象 |
| data.alarm | 预警信息数组，通常为空 |
异常与错误处理
常见 HTTP 状态码
| 状态码 | 含义 | 排查方向 |
|---|---|---|
| 200 | 正常，但需检查业务 code 是否非 0 | 解析 JSON 的业务码 |
| 401 | 未授权或密钥错误 | 检查 API Key 是否正确、是否过期 |
| 429 | 请求频率超过 QPS 限制 | 增加请求间隔，或实现本地排队与重试 |
| 5xx | 服务端异常 | 适当等待后重试，若持续则联系服务商 |
业务错误码（code 字段）
| code | msg | 原因 | 处理方式 |
|---|---|---|---|
| 1001 | 参数缺失 | 缺少必填字段 province/city | 校验请求参数完整性 |
| 1002 | 地区不存在 | 省市名无法匹配数据库 | 提示用户检查名称或提供候选 |
| 1003 | 密钥不可用 | API Key 无效或已超出额度 | 检查密钥或等待额度重置 |
工程化封装：Python 示例
现以一个WeatherClient类为例，将 curl 的调用思想转化为具备健壮性的代码封装。
import requests import logging from time import sleep from typing import Optional, Dict, Any logger = logging.getLogger("WeatherClient") class WeatherClient: """腾讯天气客户端封装""" BASE_URL = "https://v1.apizero.cn/api/tencent-weather" DEFAULT_TIMEOUT = 10 # 秒 MAX_RETRIES = 3 RETRY_BACKOFF = 1.5 # 重试间隔倍数 def __init__(self, api_key: Optional[str] = None): self.api_key = api_key self.session = requests.Session() # 每次请求都带上 Content-Type self.session.headers.update({"Content-Type": "application/json"}) if api_key: # 两种鉴权方式任选其一，这里使用 Authorization Headers self.session.headers["Authorization"] = f"Bearer {api_key}" def _do_request(self, payload: Dict[str, str]) -> requests.Response: """执行 POST 请求，包含重试逻辑""" for attempt in range(self.MAX_RETRIES): try: resp = self.session.post( self.BASE_URL, json=payload, timeout=self.DEFAULT_TIMEOUT, ) resp.raise_for_status() # 触发 HTTP 层面的错误 return resp except requests.exceptions.Timeout: logger.warning(f"请求超时，剩余重试次数 {self.MAX_RETRIES - attempt - 1}") except requests.exceptions.ConnectionError as e: logger.error(f"连接错误: {e}") except requests.exceptions.HTTPError as e: status = e.response.status_code # 4xx 错误除了 429 通常不应重试 if 400 <= status < 500 and status != 429: raise logger.warning(f"HTTP {status}，剩余重试次数 {self.MAX_RETRIES - attempt - 1}") if attempt < self.MAX_RETRIES - 1: sleep(self.RETRY_BACKOFF ** attempt) raise RuntimeError(f"请求失败，已重试 {self.MAX_RETRIES} 次") def get_weather(self, province: str, city: str, county: Optional[str] = None) -> Dict[str, Any]: """ 查询天气 :param province: 省/直辖市 :param city: 市 :param county: 区县（可选） :return: 解析后的 JSON data 字段 """ payload = {"province": province, "city": city} if county: payload["county"] = county response = self._do_request(payload) result = response.json() if result.get("code") != 0: logger.error(f"业务错误: code={result.get('code')}, msg={result.get('msg')}, request_id={result.get('request_id')}") raise ValueError(f"天气查询失败: {result.get('msg')}") return result["data"] # 使用示例 if __name__ == "__main__": logging.basicConfig(level=logging.INFO) client = WeatherClient(api_key="your-api-key-here") try: data = client.get_weather("广东", "深圳", "南山") print(f"当前温度: {data['observe']['temperature']}°C") print(f"空气质量: {data['air']['quality']}") print(f"建议衣着: {[i['detail'] for i in data['life_index'] if i['key'] == 'clothes'][0]}") except Exception as e: print(f"异常: {e}")封装要点说明
- 连接复用：使用
requests.Session()保持连接池，避免每次请求都新建 TCP 连接。 - 超时控制：
timeout=10防止接口异常时进程卡死。 - 日志记录：记录每次失败和成功的关键信息（request_id 用于排查）。
- 业务错误码校验：不仅检查 HTTP 状态码，还解析 JSON 中的
code字段，确保业务逻辑正确。 - 类型提示：使用 Python 类型注解，提升代码可维护性。
工程化注意事项（生产环境补充）
- 频率控制：QPS 上限为 10，若多个微服务共享同一个 API Key，需在客户端做本地限流（如令牌桶），避免触发 429。
- 缓存策略：天气数据变化不算频繁（实时温度除外），可按需对 hourly/daily 预报缓存 10-30 分钟，减少 API 调用次数。
- 参数标准化：用户输入的省市名可能存在空格、简繁混用，建议先进行标准化映射（如“深圳”->“深圳市”），或参考行政区划码。
- 限行解析：
limit.tail_number字段格式为“3和8”，可根据本地规则解析出具体数字，注意多城市限行规则差异。 - 监控与告警：对
code非 0 的响应、频繁的超时或 5xx 设置监控指标，及时发现接口或密钥问题。 - 多语言封装：除 Python 外，也可用 Java（OkHttp/WebClient）、Go（net/http）等实现类似的封装，核心思想一致。
参考文档
- 腾讯天气 API 官方文档：https://apizero.cn/aidocs/tencent-weather
- 原始接口规范（Markdown）：https://apizero.cn/aidocs/tencent-weather/raw.md
