从 curl 到封装：腾讯天气 API 的工程化接入指南

为什么需要从 curl 走向工程封装

在调试阶段，一条简单的curl命令就能验证接口是否通畅、返回数据是否合理。但一旦要将天气预报、生活指数等功能集成到生产系统里，curl就远远不够了——你需要处理网络抖动时的重试、接口限流、参数校验、日志记录、响应解析异常等。本文以腾讯天气 API 为例，展示如何从原型级的curl一步一步过渡到一个可维护、可扩展的工程封装。

接口能力与适用场景

腾讯天气 API 提供了基于中文省市名的天气数据查询，无需经纬度坐标。主要能力包括：

• 实时天气：温度、湿度、风向风力、天气现象、更新时间
• 空气质量：AQI、PM2.5、PM10、质量等级
• 未来 7 天预报：每日最高/最低温度
• 24 小时逐时预报：每个整点的温度和天气
• 23 项生活指数：穿衣、紫外线、洗车、运动等
• 日出日落时间：每日的具体时刻
• 机动车限行：根据城市及区县返回限行尾号

典型使用场景：

• 智能家居控制面板显示室外天气
• 旅游 App 提供目的地未来一周天气概览
• 物流调度系统结合天气与限行规划路线
• 个人助手自动推送当日穿衣建议和限行提醒

请求参数与鉴权

接口基本信息

Header 参数

注意：官方文档中也可使用X-API-Key头部传递密钥，两种方式等价，选择其一即可。

Body 参数（JSON）

{ "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": [] } }

关键字段说明

异常与错误处理

常见 HTTP 状态码

业务错误码（code 字段）

工程化封装：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}")

封装要点说明

1. 连接复用：使用requests.Session()保持连接池，避免每次请求都新建 TCP 连接。
2. 超时控制：timeout=10防止接口异常时进程卡死。
3. 日志记录：记录每次失败和成功的关键信息（request_id 用于排查）。
4. 业务错误码校验：不仅检查 HTTP 状态码，还解析 JSON 中的code字段，确保业务逻辑正确。
5. 类型提示：使用 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