在Django+DRF的CreateAPIView中配置Celery实现延迟任务

我已经为Django项目配置了Celery处理后台任务，但创建Event时，无法实现后台60秒睡眠延迟且不阻塞请求。相关代码如下：

models.py #

class Event(models.Model):
    # Fields description

    def __str__(self) -> str:
        return self.title

    class Meta():
        verbose_name = 'Мероприятие'
        verbose_name_plural = 'Мероприятия'

serializers.py #

class EventCreateSerializer(serializers.ModelSerializer):
    organizations = serializers.PrimaryKeyRelatedField(
        queryset=Organization.objects.all(),
        many=True,
        required=True
    )

    class Meta:
        model = Event
        fields = ('title', 'description', 'organizations', 'date', 'image')

views.py #

class EventCreateView(CreateAPIView):
    """
    创建新的活动。
    """
    serializer_class = EventCreateSerializer
    queryset = Event.objects.all()
    # permission_classes = [IsAuthenticated,]

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

tasks.py #

from celery import shared_task
from time import sleep
from .models import Event

@shared_task
def async_task(event_id):
    sleep(60)
    event = Event.objects.get(pk=event_id)
    print(f'Async task completed for event with ID {event_id}')

目前EventCreateView能正常运行，但任务似乎会立即执行，对象创建时没有预期的延迟。需要做哪些修改才能正确实现60秒的睡眠延迟？

问题核心是你没有触发Celery异步任务——当前仅定义了任务，但未在Event创建后将任务提交到Celery队列。按以下步骤修改：

1. 在views.py中导入并调用Celery任务 #

修改EventCreateView，重写perform_create方法，在Event保存后提交异步任务：

from .tasks import async_task  # 导入任务

class EventCreateView(CreateAPIView):
    """
    创建新的活动。
    """
    serializer_class = EventCreateSerializer
    queryset = Event.objects.all()
    # permission_classes = [IsAuthenticated,]

    def perform_create(self, serializer):
        # 先保存Event对象到数据库
        event = serializer.save()
        # 用delay()提交任务到Celery队列，而非直接调用async_task(event.id)
        async_task.delay(event.id)

    def create(self, request, *args, **kwargs):
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        headers = self.get_success_headers(serializer.data)
        return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)

2. 关键注意点 #

• 必须使用async_task.delay(event.id)：直接调用async_task(event.id)会在当前请求线程执行，导致请求阻塞；delay()会将任务发送到Celery队列，由后台worker异步执行。
• 确保Celery Worker正在运行：在项目根目录执行启动命令：

celery -A 你的项目名称 worker --loglevel=info

3. 验证Celery配置正确性 #

检查settings.py中的Celery broker配置（以Redis为例）：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

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