如何在Django APIView中追踪耗时任务并返回CSV文件？

1. 定义Report模型跟踪任务状态 #

先创建数据库模型来存储报告任务的状态与生成的CSV文件信息：

from django.db import models
from django.core.files.storage import FileSystemStorage

class Report(models.Model):
    STATUS_CHOICES = [
        ('RUNNING', 'Running'),
        ('COMPLETED', 'Completed'),
        ('FAILED', 'Failed'),
    ]
    report_id = models.CharField(max_length=255, unique=True)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='RUNNING')
    csv_file = models.FileField(storage=FileSystemStorage(location='reports/'), null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

2. 异步执行耗时任务（以Celery为例） #

generate_report4是耗时任务，必须异步执行避免阻塞API请求，这里用Celery实现：

基础Celery配置 #

# celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')
app = Celery('你的项目名')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

把generate_report4转为Celery任务 #

# tasks.py
from celery import shared_task
from .models import Report
import csv
import time  # 模拟耗时操作

@shared_task
def generate_report4_task(report_id):
    try:
        # 替换为实际的报告生成逻辑
        time.sleep(10)
        
        # 生成CSV文件
        csv_path = f'reports/{report_id}.csv'
        with open(csv_path, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['列1', '列2'])
            writer.writerow(['数据1', '数据2'])
        
        # 更新报告状态与文件路径
        report = Report.objects.get(report_id=report_id)
        report.status = 'COMPLETED'
        report.csv_file = csv_path
        report.save()
    except Exception as e:
        report = Report.objects.get(report_id=report_id)
        report.status = 'FAILED'
        report.save()
        raise e

3. 修改TriggerReportView触发异步任务 #

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
import uuid
from .models import Report
from .tasks import generate_report4_task

class TriggerReportView(APIView):
    def post(self, request):  # 建议用POST触发任务
        report_id = str(uuid.uuid4())
        # 创建初始报告记录，状态默认RUNNING
        Report.objects.create(report_id=report_id)
        # 异步启动报告生成任务
        generate_report4_task.delay(report_id)
        return Response({'report_id': report_id}, status=status.HTTP_202_ACCEPTED)

4. 实现GetReportView的状态检查与CSV返回 #

from django.http import FileResponse, Http404
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import Report

class GetReportView(APIView):
    def get(self, request):
        report_id = request.query_params.get('report_id')
        if not report_id:
            return Response({'error': 'report_id参数必填'}, status=status.HTTP_400_BAD_REQUEST)
        
        try:
            report = Report.objects.get(report_id=report_id)
        except Report.DoesNotExist:
            return Response({'error': '无效的report_id'}, status=status.HTTP_404_NOT_FOUND)
        
        if report.status == 'RUNNING':
            return Response({'status': 'Running'}, status=status.HTTP_200_OK)
        elif report.status == 'FAILED':
            return Response({'status': 'Failed', 'error': '报告生成失败'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
        elif report.status == 'COMPLETED':
            if not report.csv_file:
                return Response({'error': '报告已完成但未找到CSV文件'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
            # 返回CSV文件
            return FileResponse(
                open(report.csv_file.path, 'rb'),
                content_type='text/csv',
                filename=f'{report_id}.csv'
            )

关键说明 #

• 任务跟踪：通过数据库模型持久化任务状态，重启服务后状态不丢失；
• 异步执行：必须用Celery或类似异步框架处理耗时任务，避免API阻塞；
• 错误处理：捕获任务异常并更新状态为FAILED，方便前端排查问题；
• 文件存储：用Django的FileField管理CSV，可按需切换为云存储（如S3）。

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