flask 定时任务 flask-apscheduler

flask-apscheduler

将

apscheduler

移植到了

flask

应用中，使得在

flask

中可以非常方便的使用定时任务了，除此之外，它还有如下几个特性

• 根据Flask配置加载调度器配置
• 根据Flask配置加载任务调度器
• 允许指定服务器运行任务
• 提供RESTful API管理任务，也就是远程管理任务
• 为RESTful API提供认证

下载安装

pip install flask-apscheduler

基本使用

flask-apscheduler

的相关配置，我们会将它和其它扩展一起，放在应用的配置里

class Config(object):
    // 配置项
    JOBS = [
        {
            'id': 'job1',
            'func': 'run:add',
            'args': (1, 2),
            'trigger': 'interval',
            'seconds': 3
        }
    ]

    SCHEDULER_API_ENABLED = True

def add(a, b):
    print(a+b)

JOBS列表的每一个元素表示一个定时任务，列子中只有一个interval任务，表示每隔3秒运行一次函数add。func指定调用的函数，args表示传入函数的参数，trigger表示启动方式，常用的有两种，分别是trigger和cron。

上边我们设置了

SCHEDULER_API_ENABLED = True

，可以通过访问 http://127.0.0.1:5000/scheduler ，其中

scheduler

是默认的

RESTful API

前缀

通过查看源码，可以发现flask-apscheduler提供了以下的接口

def _load_api(self):
    """
    Add the routes for the scheduler API.
    """
    self._add_url_route('get_scheduler_info', '', api.get_scheduler_info, 'GET')
    self._add_url_route('add_job', '/jobs', api.add_job, 'POST')
    self._add_url_route('get_job', '/jobs/<job_id>', api.get_job, 'GET')
    self._add_url_route('get_jobs', '/jobs', api.get_jobs, 'GET')
    self._add_url_route('delete_job', '/jobs/<job_id>', api.delete_job, 'DELETE')
    self._add_url_route('update_job', '/jobs/<job_id>', api.update_job, 'PATCH')
    self._add_url_route('pause_job', '/jobs/<job_id>/pause', api.pause_job, 'POST')
    self._add_url_route('resume_job', '/jobs/<job_id>/resume', api.resume_job, 'POST')
    self._add_url_route('run_job', '/jobs/<job_id>/run', api.run_job, 'POST')

如果需要查看当前运行的所有定时任务，则请求http://127.0.0.1:5000/scheduler/jobs即可。

interval启动方式

interval表示间隔启动，在interval方式中，使用seconds配置间隔多久启动一次，单位是秒。

cron启动方式

cron表示定时启动

class Config(object):
    JOBS = [
        {
            'id': 'job1',
            'func': 'scheduler:task',
            'args': (1, 2),
            'trigger': 'cron',
            'day': '*',
            'hour': '13',
            'minute': '16',
            'second': '20'
        }
    ]
    SCHEDULER_API_ENABLED = True
 
 
def task(a, b):
    print(str(datetime.datetime.now()) + ' execute task ' + '{}+{}={}'.format(a, b, a + b))

该配置项则表示每天的13点16分20秒启动一次。*表示全部。

有关常用的cron配置有：

• day - 表示天
• hour - 表示小时
• minute - 表示分钟
• second - 表示秒
• week - 周
• day_of_week - 星期几，通过查找源码查询星期一到星期天分别为：- ['mon', 'tue', 'wed', 'thu', 'fri', 'sat', 'sun']

使用装饰器定时启动任务

from flask import Flask
from flask_apscheduler import APScheduler
import datetime
 
 
class Config(object):
    SCHEDULER_API_ENABLED = True
 
 
scheduler = APScheduler()
 
 
# interval examples
@scheduler.task('interval', id='do_job_1', seconds=30, misfire_grace_time=900)
def job1():
    print(str(datetime.datetime.now()) + ' Job 1 executed')

表示每隔30秒调用一次job1函数。