如何在Django中借助Celery实现三个功能：文件存储、Pandas处理、结果展示

第一步：完成Django集成Celery的基础配置 #

安装依赖 #

pip install celery redis pandas openpyxl

新增Celery配置文件 #

在项目根目录（和settings.py同级）新建celery.py：

import os
from celery import Celery

# 替换成你实际的Django项目名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')
app = Celery('你的项目名')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

在项目根目录的__init__.py添加如下内容：

from .celery import app as celery_app
__all__ = ('celery_app',)

在settings.py末尾添加Celery配置：

CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'

扩展Document模型 #

给文件模型新增状态、处理结果存储字段，执行makemigrations和migrate生效：

# models.py
from django.db import models

class Document(models.Model):
    docfile = models.FileField(upload_to='uploads/')
    # 新增字段
    status = models.CharField(max_length=20, default='pending', choices=[
        ('pending','待处理'),('processing','处理中'),('success','成功'),('failed','失败')
    ])
    html_result = models.TextField(blank=True, null=True)
    processed_file = models.FileField(upload_to='processed/', blank=True, null=True)
    error_msg = models.TextField(blank=True, null=True)
    upload_time = models.DateTimeField(auto_now_add=True)

第二步：定义Celery异步任务 #

在你的APP目录下新建tasks.py：

from celery import shared_task
from .models import Document
import pandas as pd
import io
from django.core.files.base import ContentFile

@shared_task
def process_excel_task(document_id):
    doc = Document.objects.get(id=document_id)
    doc.status = 'processing'
    doc.save()
    try:
        # 读取上传的Excel文件
        dfs = pd.read_excel(doc.docfile.path, sheet_name=None, index_col=[0])
        output = io.BytesIO()
        writer = pd.ExcelWriter(output)
        all_html = ""
        
        for name, df in dfs.items():
            # 这里替换成你实际的Pandas处理逻辑
            done = df
            done.to_excel(writer, sheet_name=name)
            all_html += f"<h3>工作表：{name}</h3>" + done.to_html()
        
        # 保存处理后的Excel文件
        output.seek(0)
        doc.processed_file.save(f"processed_{doc.docfile.name}", ContentFile(output.read()))
        # 保存HTML渲染结果
        doc.html_result = all_html
        doc.status = 'success'
        doc.save()
        return True
    except Exception as e:
        doc.status = 'failed'
        doc.error_msg = str(e)
        doc.save()
        return False

第三步：改造视图逻辑 #

# views.py
from django.shortcuts import render, redirect, get_object_or_404
from django.http import HttpResponse
from .forms import DocumentForm
from .models import Document
from .tasks import process_excel_task

def save_exls(request):
    if request.method == 'POST':
        form = DocumentForm(request.POST, request.FILES)
        if form.is_valid():
            newdoc = form.save()
            # 触发异步处理任务，仅传递可序列化的文档ID
            process_excel_task.delay(newdoc.id)
            return redirect('html_exls', doc_id=newdoc.id)
    else:
        form = DocumentForm()
    documents = Document.objects.all().order_by('-upload_time')
    return render(request, 'list.html', {'documents': documents, 'form': form})

def html_exls(request, doc_id):
    doc = get_object_or_404(Document, id=doc_id)
    return render(request, 'list.html', {'form': DocumentForm(), 'doc': doc})

def download_processed_excel(request, doc_id):
    doc = get_object_or_404(Document, id=doc_id)
    if doc.status != 'success' or not doc.processed_file:
        return HttpResponse("文件未处理完成", status=400)
    response = HttpResponse(doc.processed_file.read(), content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
    response['Content-Disposition'] = f'attachment; filename=processed_{doc.docfile.name}'
    return response

配置路由urls.py：

from django.urls import path
from . import views

urlpatterns = [
    path('upload/', views.save_exls, name='save_exls'),
    path('result/<int:doc_id>/', views.html_exls, name='html_exls'),
    path('download/<int:doc_id>/', views.download_processed_excel, name='download_excel'),
]

第四步：修改前端模板 #

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>Django site</title>
</head>
<body>
    <form action="{% url "save_exls" %}" method="post" enctype="multipart/form-data">
        {% csrf_token %}
        <p>{{ form.non_field_errors }}</p>
        <p>{{ form.docfile.label_tag }} {{ form.docfile.help_text }}</p>
        <p>
            {{ form.docfile.errors }}
            {{ form.docfile }}
        </p>
        <p><input type="submit" value="上传并处理"/></p>
    </form>
    <br/>
    {% if doc %}
        <div>
            <h4>处理状态：{{ doc.get_status_display }}</h4>
            {% if doc.status == 'failed' %}
                <p style="color: red;">处理失败：{{ doc.error_msg }}</p>
            {% elif doc.status == 'success' %}
                <p><a href="{% url 'download_excel' doc.id %}">下载处理后的Excel文件</a></p>
                <div>{{ doc.html_result|safe }}</div>
            {% else %}
                <p>文件正在处理中，可刷新页面查看最新状态</p>
                <!-- 可选：开启3秒自动刷新 -->
                <meta http-equiv="refresh" content="3">
            {% endif %}
        </div>
    {% endif %}
</body>
</html>

第五步：启动服务 #

1. 先启动Redis服务
2. 启动Celery worker：
   • Linux/Mac命令：celery -A 你的项目名 worker --loglevel=info
   • Windows命令：celery -A 你的项目名 worker --loglevel=info --pool=solo
3. 启动Django服务即可正常使用

注意事项：不要在Celery任务中直接传递HttpRequest、UploadedFile这类无法序列化的对象，统一传递数据库主键、文件路径这类可序列化参数

内容的提问来源于stack exchange，提问作者project.py