flask_apscheduler 定时任务框架

 flask 框架下定时任务通常使用 flask_apscheduler 第三方库。flask_apscheduler 功能非常强大，能满足各种定时任务的需求。

 

安装

pip3 install flask_apscheduler

使用方法

1.下面这段 example 是官网上的，我加了些注释，方便学习。

from flask import Flask
from flask_apscheduler import APScheduler


class Config(object):
    JOBS = [
        {
            'id': 'job1',                # 一个标识
            'func': '__main__:job1',     # 指定运行的函数 
            'args': (1, 2),              # 传入函数的参数
            'trigger': 'interval',       # 指定 定时任务的类型
            'seconds': 10                # 运行的间隔时间
        }
    ]

    SCHEDULER_API_ENABLED = True


def job1(a, b):                          # 运行的定时任务的函数
    print(str(a) + ' ' + str(b))


if __name__ == '__main__':
    app = Flask(__name__)                 # 实例化flask
    app.config.from_object(Config())      # 为实例化的 flask 引入配置 

    scheduler = APScheduler()                  # 实例化 APScheduler
    # it is also possible to enable the API directly
    # scheduler.api_enabled = True
    scheduler.init_app(app)                    # 把任务列表放入 flask
    scheduler.start()                          # 启动任务列表
 
    app.run()                                  # 启动 flask

 2.多个定时任务

 Config 类的属性 JOBS 是列表，可以定义多个定时任务，如下

from flask import Flask
from flask_apscheduler import APScheduler


class Config(object):
    JOBS = [
        {
            'id': '1',
            'func': '__main__:job1',
            'args': (1, 2),
            'trigger': 'interval',
            'seconds': 10
        },
        {
            'id': '2',
            'func': '__main__:job2',
            'trigger': 'interval',
            'seconds': 10
        }
    ]

    SCHEDULER_API_ENABLED = True


def job1(a, b):
    print(str(a) + ' ' + str(b))

def job2():
    print('Hello world!')


if __name__ == '__main__':
    app = Flask(__name__)
    app.config.from_object(Config())

    scheduler = APScheduler()
    # it is also possible to enable the API directly
    # scheduler.api_enabled = True
    scheduler.init_app(app)
    scheduler.start()

    app.run()

触发器类型

当你开始定时任务时，需要为定时策略选择一个触发器（设置  class Config 中 trigger 的值）。flask_apscheduler 提供了三种类型的触发器。

• date  一次性指定固定时间，只执行一次
• interval   间隔调度，隔多长时间执行一次
• cron  指定相对时间执行，比如：每月1号、每星期一执行

一、date  最基本的一种调度，指定固定时间，只执行一次

• run_date（str）– 精确时间

class Config(object):
    JOBS = [
        {
            'id': 'job1',                
            'func': '__main__:job1',          
            'args': (1, 2),              
            'trigger': 'date',                        # 指定任务触发器 date 
            'run_date': '2020-7-23 16:50:00'          # 指定时间 2020-7-23 16:50:00 执行
        }
    ]

    SCHEDULER_API_ENABLED = True

二、interval  通过设置 时间间隔 来运行定时任务

• weeks (int) – 间隔几周
• days (int) – 间隔几天
• hours (int) – 间隔几小时
• minutes (int) – 间隔几分钟
• seconds (int) – 间隔多少秒
• start_date (datetime|str) – 开始日期
• end_date (datetime|str) – 结束日期

class Config(object):
    JOBS = [
        {
            'id': 'job1',                
            'func': '__main__:job1',          
            'args': (1, 2),              
            'trigger': 'interval',                        # 指定任务触发器 interval
            'hours': 5                                     # 每间隔5h执行
        }
    ]

    SCHEDULER_API_ENABLED = True

三 、cron 通过设置 相对时间 来运行定时任务

• year (int|str) – 年，4位数字
• month (int|str) – 月 (范围1-12)
• day (int|str) – 日 (范围1-31)
• week (int|str) – 周 (范围1-53)
• day_of_week (int|str) – 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
• hour (int|str) – 时 (范围0-23)
• minute (int|str) – 分 (范围0-59)
• second (int|str) – 秒 (范围0-59)
• start_date (datetime|str) – 最早开始日期(包含)
• end_date (datetime|str) – 最晚结束时间(包含)

class Config(object):
    JOBS = [
        {
            'id': 'job1',                
            'func': '__main__:job1',          
            'args': (1, 2),              
            'trigger': 'cron',                            # 指定任务触发器 cron
            'day_of_week': 'mon-fri',              # 每周1至周5早上6点执行 
            'hour': 6,
            'minute': 00                                    
        }
    ]

    SCHEDULER_API_ENABLED = True

设置时区

 开发好定时脚本后部署到 K8S，发现报错 Timezone offset does not match system offset: 0 != 28800. Please, check your config files. 

本地运行好好的，为什么服务器上就报错了呢？

原因是系统的时区和代码运行的时区不一致导致的。解决方法是在初始化 APScheduler() 的时候加上时区：BackgroundScheduler(timezone="Asia/Shanghai")

'''...'''
from apscheduler.schedulers.background import BackgroundScheduler

class config:
    JOBS = ['''...''']
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'

'''...'''

if __name__ == '__main__':
    scheduler = APScheduler(BackgroundScheduler(timezone="Asia/Shanghai"))
    scheduler.init_app(app)
    scheduler.start()
    app.run()