Django实现CSV异步处理与结果邮件推送需求问询

Got it, let's fix that frustrating timeout problem by moving all the slow work to background tasks! Here's how you can refactor your code to give users an immediate confirmation while their request processes behind the scenes.

Step 1: Set Up Celery (Async Task Queue) #

First, we'll use Celery—Django's go-to tool for async tasks. Install it along with Redis (our task broker):

pip install celery redis

Configure Celery in Your Project #

1. Create a celery.py file in your project root:

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')

app = Celery('your_project_name')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

2. Add these settings to your settings.py:

CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Use your Redis server URL
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

3. Start the Celery worker (keep this running in a separate terminal):

celery -A your_project_name worker --loglevel=info

Step 2: Move Heavy Logic to an Async Task #

Create a tasks.py file in your app directory, and transfer all the slow operations (CSV parsing, DB queries, email sending) into a Celery task:

from celery import shared_task
from django.core.mail import EmailMessage
import csv
from io import StringIO
import os
import logging
from .models import nummodel
from django.conf import settings

logger = logging.getLogger(__name__)

@shared_task
def process_bulk_search(temp_file_path, recipient_email):
    try:
        # Parse the uploaded CSV file
        with open(temp_file_path, 'r', encoding='utf-8') as csvfile:
            spamreader = csv.reader(csvfile, dialect='excel')
            next(spamreader)  # Skip header row
            number_list = [row[0].upper() for row in spamreader]

        # Fetch matching data from database
        data_list = nummodel.objects.filter(number__in=number_list)

        # Generate the output CSV
        output_csv = StringIO()
        writer = csv.writer(output_csv)
        writer.writerow(['a', 'b'])
        for data in data_list:
            writer.writerow([data.a, data.b])

        # Send the final report via email
        message = EmailMessage(
            "Your Bulk Search Report",
            "Your request has been processed! Find your report attached below.",
            "email@gmail.com",
            [recipient_email]
        )
        message.attach('invoice.csv', output_csv.getvalue(), 'text/csv')
        message.send()

        logger.info(f"Successfully sent report to {recipient_email}")
    
    except Exception as e:
        logger.error(f"Error processing bulk search: {str(e)}")
        # Optional: Notify user if something goes wrong
        error_msg = EmailMessage(
            "Bulk Search Failed",
            f"Sorry, we hit an issue processing your request: {str(e)}",
            "email@gmail.com",
            [recipient_email]
        )
        error_msg.send()
    
    finally:
        # Clean up the temporary file
        if os.path.exists(temp_file_path):
            os.remove(temp_file_path)

Step 3: Update Your View to Return Immediate Feedback #

Modify your view to save the uploaded file temporarily, trigger the async task, and send the user a confirmation right away:

from django.shortcuts import render
from django.contrib import messages
import os
from django.conf import settings
from .tasks import process_bulk_search

def bulkserachpro(request):
    if request.method == 'POST' and request.FILES.get('searchnum'):
        uploaded_file = request.FILES['searchnum']
        
        # Create a temp directory to store the uploaded file
        temp_dir = os.path.join(settings.MEDIA_ROOT, 'temp_bulk_searches')
        os.makedirs(temp_dir, exist_ok=True)
        temp_file_path = os.path.join(temp_dir, uploaded_file.name)

        # Save the uploaded file to temp location
        with open(temp_file_path, 'wb+') as destination:
            for chunk in uploaded_file.chunks():
                destination.write(chunk)
        
        # Get recipient email (adjust this to use the user's actual email, e.g., from request.user)
        recipient_email = request.user.email if request.user.is_authenticated else "emailto@gmail.com"
        
        # Trigger the async task
        process_bulk_search.delay(temp_file_path, recipient_email)

        # Send immediate success message to the user
        messages.success(request, '文件已接收，我们会尽快处理并将报告发送至您的邮箱，请留意查收！')
        return render(request, 'bulkpro.html')
    
    return render(request, 'bulkpro.html')

Key Tips #

• Keep Celery Running: The Celery worker must stay active for tasks to execute—don't close its terminal window.
• Handle Large Files: Saving the file temporarily instead of passing its content directly avoids memory overload with big CSVs.
• Personalize Emails: Replace the hardcoded recipient email with the user's actual email (e.g., from a form field or their account profile) for a better experience.
• Monitor Errors: The task includes logging to help you debug any issues that pop up during processing.

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