Django 使用定时任务检查保质期

记录了在Django 中设置定时任务的过程。

需求

检查药品管理系统的数据库中有无过期药品。药品保质期和生产日期在两张不同表上，一个药品保质期对应多个生产日期，生产日期加保质期相加得到结果。

解决

Django 框架有一个 django-crontab 库也可以定时任务，但由于它是通过运行Linux 自带的cron 命令实现的，所以在Windows 系统上不能使用，所以最终选择了apscheduler实现

  • 首先pip install apschedule 下载apscheduler模块，再pip install django-apscheduler，后者是保障定时任务在Django 项目中运行的库，包括注册、新建定时任务数据库等功能
  • 在setting.py 的INSTALLED_APP中加入django-apschedule;
  • 使用manage.py迁移数据库，会生成两个表django_apscheduler_djangojob 和 django_apscheduler_djangojobexecution用于存放定时器；
  • 在一个view 中写定时任务代码，这样只要运行程序，定时器就会被唤醒
  • 初始化一个 apscheduler.schedulers.background.BackgroundScheduler() 实例，这是后台的定时器；
  • 在要实现的任务函数上使用装饰器@register_job 进行设置，刚刚初始化的定时器作为第一个参数传入，除此之外可以传入：
    • cron 选择cron风格，传入周、时、分、秒定时运行；
    • interval 选择interval，传入时、分、秒，每间隔这么久运行
  • 要实现的功能放在任务函数中
  • 使用register_event()将定时器注册，监控任务；
  • 最后定时器start() 开始运行。

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from apscheduler.schedulers.background import BackgroundScheduler  
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job

try:    
    scheduler = BackgroundScheduler()

    @register_job(scheduler, 'cron', day_of_week='mon-sun', hour='0', minute='0', second='0',id='task_time')
    def my_job():
        #从药品数据库取出保质期
        for medicine in CusMedicine.objects.all(): 
            exp = datetime.timedelta(days=30 * medicine.expiration)
            #从保质期数据库取出当前药品的生产日期
            for date in medicine.expirmng_set.all().values()：
                now = datetime.datetime.now()
                #这里注意，从mysql 中取出的日期类型是 datetime.date
                #但timedelta 是datetime.datetime 
                #所以需要先将datetime.data 转成str ，再转成datetime.datetime
                if now > datetime.datetime.strptime(str(date['pro_date']), '%Y-%m-%d') + exp:
                    expired.append(date)
    #注册定时器
    register_events(scheduler)
    #开始定时运行
    scheduler.start()

两张表的结构

django_apscheduler_djangojob
+———–+—————+——+—–+———+—————-+
| Field | Type | Null | Key | Default | Extra |
+———–+—————+——+—–+———+—————-+
| id | int(11) | NO | PRI | NULL | auto_increment |
| status | varchar(50) | NO | | NULL | |
| run_time | datetime(6) | NO | MUL | NULL | |
| duration | decimal(15,2) | YES | | NULL | |
| started | decimal(15,2) | YES | | NULL | |
| finished | decimal(15,2) | YES | | NULL | |
| exception | varchar(1000) | YES | | NULL | |
| traceback | longtext | YES | | NULL | |
| job_id | int(11) | NO | MUL | NULL | |
+———–+—————+——+—–+———+—————-+

django_apscheduler_djangojobexecution
+—————+————–+——+—–+———+—————-+
| Field | Type | Null | Key | Default | Extra |
+—————+————–+——+—–+———+—————-+
| id | int(11) | NO | PRI | NULL | auto_increment |
| name | varchar(255) | NO | UNI | NULL | |
| next_run_time | datetime(6) | YES | MUL | NULL | |
| job_state | longblob | NO | | NULL | |
+—————+————–+——+—–+———+—————-+