在Django REST Framework中定时上传本地文件至S3并返回S3路径的实现问询

我来帮你实现这个需求，下面是具体的步骤和代码示例，用到 Django REST Framework + Celery 处理延迟任务，结合 django-storages 对接 S3：

我们需要分三步完成需求：

1. 配置本地临时存储，接收文件上传
2. 用 Celery 实现延迟1小时上传到S3的异步任务
3. 设计API接口，处理上传请求并返回S3路径

1. 模型设计 #

首先定义一个模型来存储文件的本地路径、S3 URL 和上传状态：

from django.db import models

class UploadedFile(models.Model):
    # 本地临时存储路径，upload_to 指定本地存储的子目录
    local_file_path = models.FileField(upload_to='temp_uploads/')
    # 存储S3上的文件URL，初始为空
    s3_file_url = models.URLField(null=True, blank=True)
    # 标记是否已经上传到S3
    is_uploaded_to_s3 = models.BooleanField(default=False)
    # 记录文件创建时间
    created_at = models.DateTimeField(auto_now_add=True)

2. 配置Celery实现延迟任务 #

Celery 是Django常用的异步任务框架，用来处理延迟执行的上传任务。

2.1 安装依赖 #

pip install celery redis django-storages boto3

• celery：处理异步任务
• redis：作为Celery的消息队列（也可以用RabbitMQ）
• django-storages：简化Django与S3的对接
• boto3：AWS官方的Python SDK

2.2 配置Celery #

在项目根目录创建celery.py：

import os
from celery import Celery

# 设置Django配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')

app = Celery('your_project')
# 从Django settings读取Celery配置，前缀为CELERY_
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现所有app中的tasks.py
app.autodiscover_tasks()

在项目的__init__.py中添加：

from .celery import app as celery_app

__all__ = ('celery_app',)

在settings.py中添加Celery配置：

# Celery配置
CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Redis地址
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TIMEZONE = 'Asia/Shanghai'  # 替换为你的时区

3. 配置S3存储 #

在settings.py中添加S3相关配置：

INSTALLED_APPS = [
    # ... 其他app
    'storages',  # 添加django-storages
]

# AWS S3配置
AWS_ACCESS_KEY_ID = '你的AWS Access Key'
AWS_SECRET_ACCESS_KEY = '你的AWS Secret Key'
AWS_STORAGE_BUCKET_NAME = '你的S3存储桶名称'
AWS_S3_REGION_NAME = '你的存储桶区域，比如us-east-1'
AWS_S3_CUSTOM_DOMAIN = f'{AWS_STORAGE_BUCKET_NAME}.s3.{AWS_S3_REGION_NAME}.amazonaws.com'
AWS_DEFAULT_ACL = 'public-read'  # 根据需求设置文件权限，比如private

4. 编写异步上传任务 #

在你的app目录下创建tasks.py，编写上传到S3的任务：

from celery import shared_task
from django.conf import settings
import boto3
import uuid
from .models import UploadedFile

@shared_task(autoretry_for=(Exception,), retry_backoff=300)
def upload_to_s3(file_id):
    """延迟上传文件到S3的任务"""
    try:
        uploaded_file = UploadedFile.objects.get(id=file_id)
        
        # 读取本地文件
        with open(uploaded_file.local_file_path.path, 'rb') as file_obj:
            # 初始化S3客户端
            s3_client = boto3.client(
                's3',
                aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
                aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
                region_name=settings.AWS_S3_REGION_NAME
            )
            
            # 生成唯一的S3文件键，避免重名
            file_name = uploaded_file.local_file_path.name.split("/")[-1]
            s3_key = f'uploads/{uuid.uuid4()}_{file_name}'
            
            # 上传文件到S3
            s3_client.upload_fileobj(file_obj, settings.AWS_STORAGE_BUCKET_NAME, s3_key)
            
            # 生成S3文件URL
            s3_url = f'https://{settings.AWS_S3_CUSTOM_DOMAIN}/{s3_key}'
            
            # 更新模型记录
            uploaded_file.s3_file_url = s3_url
            uploaded_file.is_uploaded_to_s3 = True
            uploaded_file.save()
            
            # 可选：上传完成后删除本地临时文件
            # uploaded_file.local_file_path.delete()
            
    except UploadedFile.DoesNotExist:
        # 如果文件记录不存在，直接返回
        return

• autoretry_for：指定遇到异常时自动重试
• retry_backoff：重试间隔时间（这里是5分钟）

5. 编写序列化器和视图 #

5.1 序列化器 #

创建serializers.py，处理文件上传和返回数据：

from rest_framework import serializers
from .models import UploadedFile

class UploadedFileSerializer(serializers.ModelSerializer):
    class Meta:
        model = UploadedFile
        fields = ['id', 's3_file_url', 'is_uploaded_to_s3', 'created_at']
    
    def to_representation(self, instance):
        """自定义返回格式，优先返回S3 URL"""
        data = super().to_representation(instance)
        if instance.is_uploaded_to_s3:
            data['file_url'] = instance.s3_file_url
        else:
            data['file_url'] = "文件正在等待上传到S3，请1小时后再查询"
        return data

5.2 视图 #

创建views.py，处理上传和查询请求：

from rest_framework import generics
from .models import UploadedFile
from .serializers import UploadedFileSerializer
from .tasks import upload_to_s3

class FileUploadView(generics.CreateAPIView):
    """处理文件上传的视图"""
    queryset = UploadedFile.objects.all()
    serializer_class = UploadedFileSerializer
    
    def perform_create(self, serializer):
        # 保存文件到本地
        instance = serializer.save()
        # 延迟1小时（3600秒）执行上传任务
        upload_to_s3.apply_async(args=[instance.id], countdown=3600)

class FileDetailView(generics.RetrieveAPIView):
    """查询文件详情的视图，返回S3 URL"""
    queryset = UploadedFile.objects.all()
    serializer_class = UploadedFileSerializer

6. 配置URL路由 #

在app的urls.py中添加路由：

from django.urls import path
from .views import FileUploadView, FileDetailView

urlpatterns = [
    path('upload/', FileUploadView.as_view(), name='file-upload'),
    path('files/<int:pk>/', FileDetailView.as_view(), name='file-detail'),
]

7. 启动服务 #

1. 启动Redis服务（如果用Redis作为消息队列）
2. 启动Celery Worker：

celery -A your_project worker --loglevel=info

3. 启动Django开发服务器：

python manage.py runserver

注意事项 #

• 确保本地temp_uploads/目录有读写权限，Django会自动创建该目录
• 生产环境中，Celery Worker需要用进程管理工具（如supervisor、systemd）来守护运行
• 可以根据需求调整任务的重试次数和间隔，比如retry_kwargs={'max_retries': 3}
• 如果不需要保留本地文件，上传到S3后可以调用uploaded_file.local_file_path.delete()删除

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