如何在Django REST Framework中创建天气CSV文件下载API端点

第一步：修改views.py，新增下载视图 #

首先导入需要的依赖，把原有重复的天气数据处理逻辑抽成公共方法，再新增专门的下载视图：

# 新增导入依赖
from django.http import FileResponse
import io

# 抽公共方法，避免两个接口重复写逻辑
def get_historical_weather_data(query_params):
    """获取并处理World Weather Online返回的天气数据"""
    url = 'https://api.worldweatheronline.com/premium/v1/past-weather.ashx'
    query_param = {
        'key': 'my_api_key',
        'q': query_params.get('location'),
        'format': 'json',    
        'tp': 24,
        'date': query_params.get('startdate'),
        'enddate':  query_params.get('enddate'),
        'includelocation': 'yes'
    }
    weather_data =  requests.get(url, params=query_param).json()
    output = len(weather_data['data']['weather'])
    final_weather = []
    for result in range(output):
        final= {
            'country': weather_data.get('data').get('nearest_area')[0].get('country')[0]['value'],
            'areaName': weather_data.get('data').get('nearest_area')[0].get('areaName')[0]['value'],                
            'date': weather_data.get('data').get('weather')[result].get('date'),
            'max_temperature': weather_data.get('data').get('weather')[result].get('maxtempC'),
            'min_temperature': weather_data.get('data').get('weather')[result].get('mintempC'),
            'indicator': weather_data.get('data').get('weather')[result].get('hourly')[0].get('weatherDesc')[0].get('value'),
            'Humidity': weather_data.get('data').get('weather')[result].get('hourly')[0].get('humidity'),
            'cloud_over': weather_data.get('data').get('weather')[result].get('hourly')[0].get('cloudcover'),
            'windspeed': weather_data.get('data').get('weather')[result].get('hourly')[0].get('windspeedKmph'),
            'precipitation': weather_data.get('data').get('weather')[result].get('hourly')[0].get('precipInches')    
        }
        final_weather.append(final)
    df_weather = pd.DataFrame(final_weather)
    return final_weather, df_weather

# 原有接口改造，调用公共方法
class HistoricalWeatherView(APIView):
    """获取指定位置的历史天气数据"""
    def get(self, request, **kwargs):
        final_weather, _ = get_historical_weather_data(request.query_params)
        return Response(final_weather, status=status.HTTP_200_OK)

# 新增下载接口视图
class HistoricalWeatherDownloadView(APIView):
    """下载历史天气CSV文件"""
    def get(self, request, **kwargs):
        _, df_weather = get_historical_weather_data(request.query_params)
        # 将CSV写入内存缓冲区，不需要存到服务器磁盘
        buffer = io.BytesIO()
        # utf-8-sig编码兼容Excel打开不乱码，不需要可改为utf-8
        df_weather.to_csv(buffer, index=False, encoding='utf-8-sig')
        buffer.seek(0)
        # 构造文件响应，触发浏览器自动下载
        response = FileResponse(buffer, as_attachment=True, filename='historical_weather.csv')
        return response

第二步：配置urls.py路由 #

在你的项目/应用的urls.py中新增下载接口的路由映射：

from django.urls import path
from .views import HistoricalWeatherView, HistoricalWeatherDownloadView

urlpatterns = [
    # 原有接口的路由，保留不变
    path('api/weather/history/', HistoricalWeatherView.as_view(), name='historical_weather'),
    # 新增下载接口路由
    path('api/weather/history/download/', HistoricalWeatherDownloadView.as_view(), name='historical_weather_download'),
]

注意事项 #

• 下载接口和原有接口参数一致，需要携带location、startdate、enddate三个查询参数才能正常返回数据
• 如果接口有身份校验/权限控制，记得给下载视图也加上对应的权限类配置

内容的提问来源于stack exchange，提问作者femiir