python uses celery to realize order timeout cancellation

In this article, we share the specific code of celery to realize the cancellation of order timeout for your reference. The specific contents are as follows

Instructions on the use of timed tasks in Celery official documents

Project directory structure

We need to add a new task directory, such as order:

celey_tasks/
(config. py)
_ _ init _ _. py
(main. py)
(order/)
_ _ init _ _. py
tasks. py

In main. py, register the task directory "Note that we will use the model processing of django later, so we must introduce the configuration of django"

import os

from celery import Celery

# 1.  Create a sample object 
app = Celery("luffy")

# 2.  Load configuration 
app.config_from_object("celery_tasks.config")
# 3.  Register task [ Automatically search and load tasks ]
#  Parameter must be 1 List, each of which 1 Each task is the path name of the task 
# app.autodiscover_tasks([" Mission 1"," Mission 2"])
app.autodiscover_tasks(["celery_tasks.order"])

# 4.  Run under the terminal celery Command start celery
# celery -A  Main program  worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info

Next, under the order task directory, create the task file tasks. py with the fixed name, code:

from celery_tasks.main import app

@app.task(name="check_order")
def check_order():
 print(" Check whether the order is overdue !!!")

Next, we need to set this task as a timed task, so we need to use the Crontab module provided by Celery itself.

In the configuration file, register the timed task:

#  Link address of task queue 
broker_url = 'redis://127.0.0.1:6379/15'
#  Link address of the result queue 
result_backend = 'redis://127.0.0.1:6379/14'

from celery.schedules import crontab
from .main import app
#  Scheduling list of timed tasks, which is used to register timed tasks 
app.conf.beat_schedule = {
 # Executes every Monday morning at 7:30 a.m.
 'check_order_outtime': {
 #  Tasks scheduled this time 
 'task': 'check_order', #  The task name here must come first main.py Register in 
 #  Scheduling period of timed tasks 
 # 'schedule': crontab(minute=0, hour=0), #  Every morning 00:00
 'schedule': crontab(), #  Per minute 
 # 'args': (16, 16), #  Note: The task is to 1 Function, so if there are parameters, you need to pass 
 },
}

Next, we can restart Celery and enable the timed task scheduler of Celery

First, under the terminal, run the timing task program of celery, and the following commands:

celery -A celery_tasks.main beat # ycelery.main  Yes celery Main application file of 

Then create a new terminal and run the following command, which must be specified first:

celery -A celery_tasks.main worker --loglevel=info

Timing task

After the above tests, we only need to modify the above task function to determine whether the modification order timed out.

To complete the task function of the order, if the model operation of django framework needs to be called, configuration loading and initialization must be carried out for django framework.
main. py, code

import os

from celery import Celery

# 1.  Create a sample object 
app = Celery("luffy")

#  Put celery And django Assemble, identify, and load django Configuration file for 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')

#  In the current clery Start in django Framework, right django The framework is initialized 
import django
django.setup()

# 2.  Load configuration 
app.config_from_object("celery_tasks.config")
# 3.  Register task [ Automatically search and load tasks ]
#  Parameter must be 1 List, each of which 1 Each task is the path name of the task 
# app.autodiscover_tasks([" Mission 1"," Mission 2"])
app.autodiscover_tasks(["celery_tasks.sms","celery_tasks.order"])

# 4.  Run under the terminal celery Command start celery
# celery -A  Main program  worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info

Note that because the time zone configuration is in django, we also need to modify the time zone configuration in django framework configuration.

Implementation of task code tasks. py:

from celery_tasks.main import app
from orders.models import Order
from datetime import datetime
from django.conf import settings
@app.task(name="check_order")
def check_order():
 #  Find out all orders that have timed out 
 #  Timeout condition:   Current time  > ( Order generation time  +  Timeout time ) =====>>>> ( Current time  -  Timeout time ) >  Order generation time 
 now = datetime.now().timestamp()
 timeout_number = now - settings.ORDER_TIMEOUT
 timeout = datetime.fromtimestamp(timeout_number)
 timeout_order_list = Order.objects.filter(order_status=0, created_time__lte=timeout)
 for order in timeout_order_list:
 order.order_status = 3
 order.save()

Configuration file, settings/dev. py, code:

#  Set the time when the order timeout timeout [ Unit :  Seconds ]
ORDER_TIMEOUT = 12 * 60 * 60