当前位置： 首页 > news >正文

5分钟搞定：用Python+Flask快速搭建天气预报API服务（附完整代码）

news 2026/6/10 21:02:06

从零构建高可用天气API服务：Python Flask实战指南

天气预报数据在现代应用中扮演着重要角色，从出行App到智能家居系统都可能需要实时天气信息。本文将带你用Python Flask框架构建一个功能完备的天气API服务，不仅包含基础数据获取，还会实现缓存优化、限流保护和文档自动化等生产级功能。

1. 项目架构设计

在开始编码前，我们需要明确服务的核心组件。一个完整的天气API服务应该包含以下模块：

  • 数据获取层：负责与第三方天气API交互
  • 业务逻辑层：处理数据转换和业务规则
  • API接口层：对外提供RESTful接口
  • 辅助功能层：包括缓存、限流、日志等

技术选型方面，我们使用：

  • Flask：轻量级Web框架
  • Redis：缓存和限流存储
  • Celery：异步任务队列
  • Swagger：API文档生成

提示：生产环境建议使用gunicorn或uWSGI作为WSGI服务器，而不是直接运行Flask开发服务器

2. 基础服务搭建

首先创建项目结构：

mkdir weather-api && cd weather-api python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows pip install flask redis celery python-dotenv

创建基础Flask应用（app.py）：

from flask import Flask, jsonify import os app = Flask(__name__) app.config.from_mapping( SECRET_KEY=os.getenv('SECRET_KEY', 'dev'), REDIS_URL=os.getenv('REDIS_URL', 'redis://localhost:6379/0') ) @app.route('/health') def health_check(): return jsonify({'status': 'healthy'}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

3. 集成天气数据接口

我们使用和风天气作为数据源，首先封装API客户端：

# services/weather.py import requests from datetime import datetime, timedelta from flask import current_app class WeatherService: BASE_URL = 'https://devapi.qweather.com/v7/weather' def __init__(self, api_key): self.api_key = api_key self.cache_expire = timedelta(hours=1) def get_current_weather(self, location): cache_key = f'current_{location}' cached = self._get_cache(cache_key) if cached: return cached url = f'{self.BASE_URL}/now?location={location}&key={self.api_key}' response = requests.get(url) data = response.json() if data.get('code') == '200': self._set_cache(cache_key, data) return data raise ValueError(f"Weather API error: {data.get('message')}") def _get_cache(self, key): # 实现Redis缓存获取 pass def _set_cache(self, key, value): # 实现Redis缓存设置 pass

4. 实现核心API端点

创建主要API路由（api/weather.py）：

from flask import Blueprint, request, jsonify from services.weather import WeatherService from utils.rate_limit import rate_limited bp = Blueprint('weather', __name__, url_prefix='/api/v1') @bp.route('/weather/current', methods=['GET']) @rate_limited(limit=60, per=300) # 每分钟60次请求 def current_weather(): location = request.args.get('location') if not location: return jsonify({'error': 'Location parameter is required'}), 400 try: weather = WeatherService(current_app.config['HEFENG_KEY']) data = weather.get_current_weather(location) return jsonify(data) except ValueError as e: return jsonify({'error': str(e)}), 500

5. 高级功能实现

5.1 Redis缓存集成

# utils/cache.py import redis from flask import current_app from datetime import timedelta import pickle class Cache: def __init__(self): self.redis = redis.from_url(current_app.config['REDIS_URL']) def get(self, key): val = self.redis.get(key) return pickle.loads(val) if val else None def set(self, key, value, expire=None): if expire is None: expire = current_app.config.get('CACHE_DEFAULT_EXPIRE', 3600) self.redis.setex(key, expire, pickle.dumps(value))

5.2 请求限流保护

# utils/rate_limit.py from functools import wraps from flask import request, jsonify import redis from datetime import timedelta def rate_limited(limit=60, per=300): def decorator(f): @wraps(f) def wrapped(*args, **kwargs): # 获取客户端IP作为标识 identifier = request.remote_addr key = f"rate_limit:{identifier}:{request.endpoint}" r = redis.from_url(current_app.config['REDIS_URL']) current = r.get(key) if current and int(current) >= limit: return jsonify({ 'error': 'Too many requests', 'retry_after': per }), 429 pipe = r.pipeline() pipe.incr(key) pipe.expire(key, per) pipe.execute() return f(*args, **kwargs) return wrapped return decorator

5.3 异步任务处理

对于耗时的天气数据预处理，可以使用Celery：

# tasks/weather.py from celery import Celery from services.weather import WeatherService celery = Celery(__name__, broker='redis://localhost:6379/1') @celery.task(bind=True) def fetch_weather_data(self, location): try: weather = WeatherService(current_app.config['HEFENG_KEY']) return weather.get_current_weather(location) except Exception as e: self.retry(exc=e, countdown=60)

6. API文档与测试

使用Flask-RESTX自动生成Swagger文档：

# api/docs.py from flask_restx import Api, Resource, fields from flask import Blueprint bp = Blueprint('api', __name__, url_prefix='/api') api = Api(bp, version='1.0', title='Weather API', description='A simple Weather API service') weather_model = api.model('Weather', { 'location': fields.String(required=True, description='City name or ID'), 'temperature': fields.String(description='Current temperature'), 'condition': fields.String(description='Weather condition') }) @api.route('/weather') class WeatherResource(Resource): @api.doc('get_current_weather') @api.marshal_with(weather_model) def get(self): """Get current weather for a location""" pass

7. 部署与优化建议

项目完成后，可以考虑以下优化方向：

  • 容器化部署：使用Docker打包应用
  • 负载均衡：Nginx反向代理多个实例
  • 监控告警：Prometheus + Grafana监控
  • 日志收集：ELK日志分析系统

一个简单的Dockerfile示例：

FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . ENV FLASK_APP=app.py ENV FLASK_ENV=production EXPOSE 5000 CMD ["gunicorn", "--bind", "0.0.0.0:5000", "app:app"]

在实际项目中，我发现Redis缓存命中率对性能影响很大。通过调整缓存过期策略，可以将API响应时间从平均800ms降低到200ms以下。另一个实用技巧是使用HTTP缓存头，让客户端缓存不常变的数据：

@app.route('/weather/current') def current_weather(): # ...获取天气数据... response = jsonify(data) response.headers['Cache-Control'] = 'public, max-age=300' # 5分钟缓存 return response
http://www.jsqmd.com/news/604809/

相关文章：

  • PHP 文件上传详解
  • 探寻温州高性价比本子源头：臻冠文具如何以实力定义行业标杆 - 2026年企业推荐榜
  • 配电网光伏储能双层优化配置模型：基于粒子群算法求解选址定容与运行调度联合优化
  • 避坑指南：若依Pro多数据源事务处理的3种正确姿势
  • 13.2W开关电源设计详解：从变压器计算到元器件选型
  • 2026深度解析：温州手工女鞋供应链五强格局与选型指南 - 2026年企业推荐榜
  • LD2410毫米波雷达UART通信库技术解析
  • Bootstrap5 表单浮动标签详解
  • 不锈钢外六角组合螺丝怎么选：河北不锈钢十字盘头组合螺丝/河北不锈钢圆柱头内六角组合螺丝/选择指南 - 优质品牌商家
  • 【源荷储再创新】小论文轻松发！基于雨流计数法的源-荷-储双层协同优化配置研究Matlab代码
  • 基于QT的跨平台串口调试工具开发实践
  • 5步搞定OpenClaw+Qwen3.5-9B：星图GPU镜像一键体验方案
  • Vue2集成cafe-ofd实现高效OFD文件预览方案
  • CH32软件I2C库：兼容Wire接口的GPIO模拟I2C解决方案
  • HR 系统怎么选？从功能、适配到性价比全维度解析
  • 基于单片机的车辆防盗系统（有完整资料）
  • 2026年十堰周岁宴酒店选择指南：深度解析五大服务商与前瞻决策路径 - 2026年企业推荐榜
  • BartOS-wifi-basic：ESP8266轻量级WiFi状态机驱动
  • 2024年通信与信号处理领域期刊投稿指南：如何根据影响因子和分区选择最适合的期刊
  • Arduino非阻塞摩尔斯电码库：嵌入式实时发送框架
  • OpenClaw节日应用：Qwen3.5-9B自动发送定制化祝福消息
  • 十堰朋友小聚餐厅全攻略：2026年精选推荐与联系指南 - 2026年企业推荐榜
  • HR 实操指南与落地方法：360 绩效评估全解析
  • 串口通信优化：FIFO与协议设计实战
  • 嵌入式系统内存管理：原理与实践技巧
  • BeautyPlus美颜相机 v7.33.1-会员功能已解锁！AI换装、手办图、热门滤镜基本都有
  • 老人健康监测（有完整资料）
  • ARM嵌入式系统内存对齐原理与实践
  • STM32 GPIO工作原理与实战应用详解
  • 2026年北京知识产权法律服务市场，这五支团队值得关注 - 2026年企业推荐榜