在数据分析和环境监测类应用中历史空气质量数据是一项出现频率很高的基础能力。无论是回溯某次污染过程、验证减排效果还是为业务报表提供环境维度字段都需要一个稳定、可解释的接口。本篇文章以历史空气质量接口为例记录从一条 curl 命令到完成基础工程封装的完整路径。适用场景与接口定位历史空气质量接口的核心能力是输入城市与年月返回该月逐日 AQI 与六项污染物浓度并附带空气质量等级、首要污染物、城市排名以及月度汇总统计。数据范围从 2013 年 1 月至当前月覆盖全国主要城市。在实际开发中这类接口通常用于以下场景环境研究辅助批量拉取特定城市、特定年份的逐日 AQI 数据用于趋势分析或模型输入。业务报表填充在月度运营报告中自动附加城市空气质量概况例如优良天数占比。数据校验与审计通过独立数据源交叉验证内部采集的空气质量数据。定时任务采集每月初自动拉取上月数据落库后供下游分析使用。接口一次请求返回整月数据而非单日数据因此拉取一年数据只需 12 次请求数据密度较高。接口能力边界在开始编码之前先明确接口的几个关键约束请求方式POST请求地址https://v1.apizero.cn/api/air-historyQPS 限制5 次/秒即单机 200ms 内不应发起第二个请求。可查范围2013 年 01 月至当前月超过范围会返回错误。数据粒度逐日数据 月度汇总不提供小时级或更细粒度。首要污染物计算按国标 HJ 633-2012 IAQI 分指数计算。上述信息以官方文档为准QPS 在工程设计时不允许打满需要预留余量。请求与鉴权接口是标准的 POST JSON 请求体。鉴权支持两种方式Header 中携带 Authorization:Bearer 你的 API KeyHeader 中携带 X-API-Key:你的 API Key两种方式选其一即可。匿名调用也可以但有额度限制同时不保证高可用。工程化项目建议统一使用 API Key 方式并把密钥放在环境变量中不要硬编码到仓库。请求体是一个 JSON 对象包含两个必填字段字段类型必填说明citystring是城市中文名例如“北京”monthstring是年月格式YYYYMM例如202503一个完整的请求体示例{ city: 北京, month: 202503 }从 curl 开始最小可用验证在开始写工程代码之前先用 curl 验证接口连通性、认证方式和返回结构这是维护复杂度最低的接入方式。curl -sS \ -X POST \ -H X-API-Key: $APIZERO_API_KEY \ -H Content-Type: application/json \ -d {city: 北京, month: 202503} \ https://v1.apizero.cn/api/air-history将$APIZERO_API_KEY替换为实际的密钥。如果使用匿名调用去掉X-API-Key这个 Header 即可。执行后如果返回 JSON 中包含code: 0说明请求成功。接下来可以用 Python 的json.tool对返回结果做格式化方便观察字段结构curl -sS -X POST \ -H X-API-Key: $APIZERO_API_KEY \ -H Content-Type: application/json \ -d {city: 北京, month: 202503} \ https://v1.apizero.cn/api/air-history | python3 -m json.toolPython 工程化封装curl 适合做一次性验证。当需要把接口纳入数据管道或业务系统时必须做好参数校验、超时控制、异常处理和调用频率管理。下面用一个 Python 类完成最基础的封装。基础请求类import os import time import requests class AirHistoryClient: 历史空气质量接口客户端 BASE_URL https://v1.apizero.cn/api/air-history def __init__(self, api_key: str | None None, timeout: float 5.0): :param api_key: API Key不传则匿名调用 :param timeout: 请求超时时间秒 self.api_key api_key or os.getenv(APIZERO_API_KEY) self.timeout timeout self.session requests.Session() def _build_headers(self) - dict: headers {Content-Type: application/json} if self.api_key: headers[X-API-Key] self.api_key return headers def fetch_month( self, city: str, month: str, retries: int 2 ) - dict: 拉取指定城市、月份的空气质量数据。 :param city: 城市中文名如 北京 :param month: 年月格式 YYYYMM如 202503 :param retries: 失败重试次数 :return: 接口返回的 data 字段 payload {city: city, month: month} for attempt in range(retries 1): try: resp self.session.post( self.BASE_URL, headersself._build_headers(), jsonpayload, timeoutself.timeout, ) resp.raise_for_status() body resp.json() if body.get(code) ! 0: raise RuntimeError(f接口业务错误: {body}) return body[data] except (requests.Timeout, requests.ConnectionError) as e: if attempt retries: raise RuntimeError(f请求失败已重试 {retries} 次: {e}) time.sleep(0.5 * (2**attempt)) raise RuntimeError(请求失败) # 不可达仅作类型提示调用方式与频率控制由于接口 QPS 限制为 5 次/秒实际工程中建议将调用频率控制在 1 次/秒以下。如果业务需要批量拉取 10 个城市的月度数据可以这样设计import time client AirHistoryClient() cities [北京, 上海, 广州, 深圳, 成都, 杭州, 武汉, 西安, 南京, 重庆] month 202503 results {} for ci, city in enumerate(cities): results[city] client.fetch_month(city, month, retries3) if ci len(cities) - 1: time.sleep(1) # 控制频率保证不会触发限流 print(f成功拉取 {len(results)} 个城市的数据)上面的循环中每次请求之间主动 sleep 1 秒实际 QPS 为 1远低于接口上限。字段级校验在进入业务逻辑之前对入参做前置校验可以提前暴露问题避免把错误请求发送到远端。import re import datetime def validate_month(month: str) - bool: 校验月份格式并判断是否在可查范围内 if not re.fullmatch(r\d{6}, month): return False year, mon int(month[:4]), int(month[4:]) current datetime.date.today() if (year, mon) (2013, 1): return False if (year, mon) (current.year, current.month): return False return True调用时先执行校验def fetch_with_validation(client: AirHistoryClient, city: str, month: str): if not validate_month(month): raise ValueError(fmonth 参数不合法或超出可查范围: {month}) if not city or not city.strip(): raise ValueError(city 不能为空) return client.fetch_month(city.strip(), month)返回字段解读接口成功时返回code: 0数据集中在data对象中。为了方便后续的数据建模需要理解三个子结构的含义。daily逐日明细daily是一个数组按日期升序排列每天一个元素。每个元素包含字段类型说明datestring日期格式YYYY-MM-DDaqiinteger当日 AQI 值levelstring空气质量等级描述如“良”primary_pollutantstring首要污染物名称rankinteger当日城市全国排名pollutantsobject六项污染物浓度含 pm25、pm10、so2、no2、co、o3{ date: 2024-12-01, aqi: 53, level: 良, primary_pollutant: 细颗粒物(PM2.5), rank: 123, pollutants: { pm2_5: 26, pm10: 55, so2: 4, no2: 35, co: 0.5, o3: 52 } }污染物浓度单位CO 为 mg/m³其余五项为 µg/m³在数据分析时需要注意单位差异。summary月度汇总summary提供整月的统计摘要核心字段如下aqi_avg月均 AQI浮点数aqi_avg_level月均 AQI 对应等级best_day当月 AQI 最低的一天含日期、AQI 值和等级worst_day当月 AQI 最高的一天days_distribution空气质量天数分布分优excellent、良good、轻度light、中度medium、重度heavy、严重severe六档summary: { aqi_avg: 58.3, aqi_avg_level: 良, best_day: { date: 2024-12-05, aqi: 28, level: 优 }, worst_day: { date: 2024-12-20, aqi: 168, level: 中度污染 }, days_distribution: { excellent: 8, good: 18, light: 3, medium: 2, heavy: 0, severe: 0 } }meta元信息meta: { city: 北京, month: 202412, month_format: 2024年12月, total_days: 31 }total_days可以用于校验daily数组长度防止数据缺失。建议在数据落库时做一致性检查如果len(daily) ! meta.total_days则视为异常数据。错误处理策略接口错误分为网络层错误、HTTP 状态错误和业务逻辑错误三种。HTTP 状态码状态码常见原因处理策略400请求体格式错误字段缺失或类型不符检查 payload 结构401未认证或 API Key 无效检查密钥是否正确403无访问权限检查调用额度404接口地址错误或数据不存在检查 URL 和 month 范围429请求频率超限增加 sleep 时间启用退避重试500服务端异常指数退避重试最多 3 次业务错误码HTTP 200 不代表业务成功需要判断code字段。业务错误示例与含义以原始文档为准。工程上应优先检查code ! 0的分支并携带request_id记录日志便于后续定位问题。if body.get(code) ! 0: log.error( API 业务错误: code%s, msg%s, request_id%s, body.get(code), body.get(msg), body.get(request_id), ) raise RuntimeError(f业务处理失败: {body.get(msg)})重试设计由于接口 QPS 上限为 5 次/秒重试策略应优先采用指数退避而非固定间隔重试避免故障恢复时所有客户端同时发起重试造成抢占式限流。import time def exponential_backoff(attempt: int, base: float 0.5, cap: float 5.0): return min(base * (2**attempt), cap)工程化注意事项数据缓存历史空气质量数据的同一天同一城市返回结果基本不变。建议在落库时按(city, month)做唯一约束重复请求直接读缓存减少接口调用量。CREATE TABLE air_quality_monthly ( id BIGINT PRIMARY KEY AUTO_INCREMENT, city VARCHAR(50) NOT NULL, month CHAR(6) NOT NULL, raw_data JSON NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uk_city_month (city, month) );日志与监控对每次请求记录结构化日志是定位问题的基础import logging log logging.getLogger(air_history) log.info( fetch air history: city%s month%s request_id%s , city, month, body.get(request_id), )对于定时采集任务还要对请求耗时、失败率、重试次数做基础监控。时间参数month参数格式为YYYYMM不是YYYY-MM也不是带T的 ISO 时间。批量脚本中建议使用统一的日期工具生成import datetime def current_month() - str: return datetime.datetime.now().strftime(%Y%m) def previous_month(dt: datetime.datetime) - str: first_day_next (dt.replace(day1) datetime.timedelta(days32)).replace(day1) last_day_prev first_day_next - datetime.timedelta(days1) return last_day_prev.strftime(%Y%m)限流规避QPS 5 意味着任何超过 5 次/秒的并发请求都可能被拒。工程上建议做三层保护客户端内置节流器保证单个进程输出 QPS ≤ 2。部署多实例时通过 Redis 分布式锁对同一接口设置全局限流。小结从 curl 到工程封装本质上是从“验证可用性”过渡到“保障稳定性”的过程。通过参数校验、超时控制、异常分类和缓存策略可以把一个简单接口变成可靠的数据源。对于历史空气质量这类数据变化频率低的接口缓存优先、按需拉取、低频调度是最务实的接入方式。参考文档历史空气质量接口文档原始 Markdown 文档