求助：Celery无法按预期更新Django模型PizzaOrder的order_status字段

模型定义 #

class PizzaOrder(models.Model):
    customer = models.ForeignKey(User, null=True, on_delete=models.CASCADE)
    total_price = models.DecimalField(max_digits=7, decimal_places=2, null=True, blank=True)
    order_status = models.CharField(max_length=20)
    created_at = models.DateTimeField(auto_now_add=True)

预期状态更新规则 #

• 订单提交后1分钟内：从Placed变为Accepted
• 提交1分钟后：从Accepted变为Preparing
• 提交3分钟后：从Preparing变为Dispatched
• 提交5分钟后：变为Delivered

现有Celery任务代码 #

from celery import shared_task
from datetime import timedelta
from django.utils import timezone
from .models import PizzaOrder

@shared_task
def update_order_status(order_id):
    try:
        order = PizzaOrder.objects.get(pk=order_id)
        current_time = timezone.now()
        
        if order.order_status == 'Placed' and current_time - order.created_at <= timedelta(minutes=1):
            order.order_status = 'Accepted'
        elif order.order_status == 'Accepted' and current_time - order.created_at > timedelta(minutes=1) and current_time - order.created_at <= timedelta(minutes=2):
            order.order_status = 'Preparing'
        elif order.order_status == 'Preparing' and current_time - order.created_at > timedelta(minutes=3) and current_time - order.created_at <= timedelta(minutes=5):
            order.order_status = 'Dispatched'
        elif order.order_status == 'Dispatched' and current_time - order.created_at > timedelta(minutes=5):
            order.order_status = 'Delivered'
        
        order.save()
    except PizzaOrder.DoesNotExist:
        pass

序列化器调用代码 #

class PizzaOrderSerializer(serializers.ModelSerializer):
    pizzas = PizzaSerializer(many=True)

    class Meta:
        model = PizzaOrder
        fields = ["customer", "pizzas", "total_price"]

    def create(self, validated_data):
        pizzas_data = validated_data.pop('pizzas')
        request = self.context.get('request')
        total_price = len(pizzas_data) * 69  # Calculate the total price based on the number of pizzas
        order_status = "Placed"
        pizza_order = PizzaOrder.objects.create(customer=request.user, total_price=total_price, order_status=order_status, **validated_data)

        update_order_status.apply_async(args=[pizza_order.id], countdown=0) 

Celery运行日志 #

[2023-09-28 02:19:29,116: INFO/MainProcess] Connected to redis://localhost:6379//
[2023-09-28 02:19:29,116: WARNING/MainProcess] /Users/kartikeysinha/Desktop/local-git-repository/git-env/lib/python3.9/site-packages/celery/worker/consumer/consumer.py:507: CPendingDeprecationWarning: The broker_connection_retry configuration setting will no longer determine
whether broker connection retries are made during startup in Celery 6.0 and above.
If you wish to retain the existing behavior for retrying connections on startup,
you should set broker_connection_retry_on_startup to True.
  warnings.warn(

[2023-09-28 02:19:29,119: INFO/MainProcess] mingle: searching for neighbors
[2023-09-28 02:19:30,133: INFO/MainProcess] mingle: all alone
[2023-09-28 02:19:30,152: INFO/MainProcess] celery@Kartikeys-MacBook-Air.local ready.
[2023-09-28 02:19:42,576: INFO/MainProcess] Task api.tasks.update_order_status[e8095710-77b5-4738-8fe8-05a12895f3eb] received
[2023-09-28 02:19:42,660: INFO/ForkPoolWorker-8] Task api.tasks.update_order_status[e8095710-77b5-4738-8fe8-05a12895f3eb] succeeded in 0.08167616600000116s: None

问题分析 #

1. 任务仅执行一次，无后续调度：当前仅在订单创建时调用一次任务，执行后不会自动触发后续状态更新。比如第一次将状态改为Accepted后，没有任务继续监控时间节点，无法触发Preparing及后续状态切换。
2. 时间区间判断错误：
   • 第二个条件中<= timedelta(minutes=2)不符合需求，正确区间应为1分钟到3分钟之间，才能触发Accepted到Preparing的切换。
   • 即使后续执行任务，错误的时间区间也会导致状态更新逻辑不触发。

修复方案 #

修改Celery任务，使其每次更新状态后，自动调度下一次状态更新任务；同时修正时间判断区间。

修改后的Celery任务代码 #

from celery import shared_task
from datetime import timedelta
from django.utils import timezone
from .models import PizzaOrder

@shared_task
def update_order_status(order_id):
    try:
        order = PizzaOrder.objects.get(pk=order_id)
        current_time = timezone.now()
        time_since_created = current_time - order.created_at
        next_countdown = None

        # 按状态和时间区间更新状态，并计算下一次任务的延迟时间
        if order.order_status == 'Placed':
            if time_since_created <= timedelta(minutes=1):
                order.order_status = 'Accepted'
                # 计算到下一个状态节点的剩余时间
                next_countdown = max(0, (timedelta(minutes=1) - time_since_created).total_seconds())
        elif order.order_status == 'Accepted':
            if timedelta(minutes=1) < time_since_created <= timedelta(minutes=3):
                order.order_status = 'Preparing'
                next_countdown = max(0, (timedelta(minutes=3) - time_since_created).total_seconds())
        elif order.order_status == 'Preparing':
            if timedelta(minutes=3) < time_since_created <= timedelta(minutes=5):
                order.order_status = 'Dispatched'
                next_countdown = max(0, (timedelta(minutes=5) - time_since_created).total_seconds())
        elif order.order_status == 'Dispatched':
            if time_since_created > timedelta(minutes=5):
                order.order_status = 'Delivered'
                # 最终状态，无需再调度任务
                next_countdown = None

        order.save()

        # 调度下一次任务（如果需要）
        if next_countdown is not None:
            update_order_status.apply_async(args=[order_id], countdown=next_countdown)
            
    except PizzaOrder.DoesNotExist:
        pass

序列化器调用代码（无需修改，保持原逻辑即可） #

class PizzaOrderSerializer(serializers.ModelSerializer):
    pizzas = PizzaSerializer(many=True)

    class Meta:
        model = PizzaOrder
        fields = ["customer", "pizzas", "total_price"]

    def create(self, validated_data):
        pizzas_data = validated_data.pop('pizzas')
        request = self.context.get('request')
        total_price = len(pizzas_data) * 69  # 按披萨数量计算总价
        order_status = "Placed"
        pizza_order = PizzaOrder.objects.create(customer=request.user, total_price=total_price, order_status=order_status, **validated_data)

        # 首次立即执行任务，后续由任务自动调度
        update_order_status.apply_async(args=[pizza_order.id], countdown=0) 
        return pizza_order

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