flask Flask-APScheduler 配置计划任务

pip install Flask-APScheduler

方法一：

• __init__.py

  from flask import Flask
  # 引用 APSchedule
  from flask_apscheduler import APScheduler
  # 引用 congfig 配置
  from config import Config, APSchedulerJobConfig
   
  app = Flask(__name__)
   
  # 定时任务，导入配置
  # APSchedulerJobConfig 就是在 config.py文件中的 类 名称。
  app.config.from_object(APSchedulerJobConfig)
   
  # 初始化Flask-APScheduler，定时任务
  scheduler = APScheduler()
  scheduler.init_app(app)
  scheduler.start()
  

• config.py

#crontab
class APSchedulerJobConfig(object):
  SCHEDULER_API_ENABLED = True
  JOBS = [
      {
        'id': 'No1', # 任务唯一ID
        'func': 'app.task_schedule:test_cron', # 执行任务的function名称，app.test 就是 app下面的`test.py` 文件，`shishi` 是方法名称。文件模块和方法之间用冒号":"，而不是用英文的"."
        'args': '', #如果function需要参数，就在这里添加
        'trigger': {
          'type': 'cron', # 类型
          # 'day_of_week': "0-6", # 可定义具体哪几天要执行
          # 'hour': '*', # 小时数
          # 'minute': '1',
          'second': '17' # "*/3" 表示每3秒执行一次，单独一个"3" 表示每分钟的3秒。现在就是每一分钟的第3秒时循环执行。
        }
      }
  ]

• task_schedule.py

import time, os


def test_cron():
    print('Job 17 executed')
    task_time = time.strftime("%Y-%m-%d %H:%M:%S")
    ll = 'echo {test1} >> time_test.txt'.format(test1=task_time)
    os.system(ll)

方案2

• __init__.py

from flask import Flask
 
app = Flask(__name__)
app.config.from_object('config.config')
 

 
from models.task_schedule import scheduler
scheduler.init_app(app)
scheduler.start()

• task_schedule.py

  import time, os
  
  from flask_apscheduler import APScheduler
  scheduler = APScheduler()
  
  
  #cron
  '''
  参数                       说明
  year            (int 或 str)      表示四位数的年份 （2019）
  month          （int\    str）    月 (范围1-12)
  day            （int\    str）    日 (范围1-31）
  week           （int\    str）    周 (范围1-53)
  day_of_week     (int\    str)    表示一周中的第几天，既可以用0-6表示也可以用其英语缩写表示
  hour            (int\    str)    表示取值范围为0-23时
  minute          (int\     str)    表示取值范围为0-59分
  second          (int\    str)    表示取值范围为0-59秒
  start_date      (datetime\    str)    表示开始时间
  end_date        (datetime\    str)    表示结束时间
  timezone        (datetime.tzinfo\    str)    表示时区取值
  '''
  
  '''interval:
  时间间隔触发器，每个一定时间间隔执行一次。
  
  | 参数 | 说明 |
  | —————————- | ———- |
  | weeks (int) | 间隔几周 |
  | days (int) | 间隔几天 |
  | hours (int) | 间隔几小时 |
  | minutes (int) | 间隔几分钟 |
  | seconds (int) | 间隔多少秒 |
  | start_date (datetime 或 str) | 开始日期 |
  | end_date (datetime 或 str) | 结束日期 |
  
  # 在 2019-4-24 00:00:00 - 2019-4-24 08:00:00 之间, 每隔两小时执行一次 alarm_job 方法
  scheduler .add_job(alarm_job, 'interval', hours=2, start_date='2019-4-24 00:00:00' , end_date='2019-4-24 08:00:00')
  '''
  
  
  '''date：
  固定日期触发器，任务只运行一次，运行完毕自动清除；若错过指定运行时间，任务不会被创建
  
  | 参数 | 说明 |
  | :——————————– | :——————- |
  | run_date (datetime 或 str) | 作业的运行日期或时间 |
  | timezone (datetime.tzinfo 或 str) | 指定时区 |
  
  例如# 在 2019-4-24 00:00:01 时刻运行一次 start_system 方法
  scheduler .add_job(start_system, 'date', run_date='2019-4-24 00:00:01', args=['text'])
  
  '''
  
  
  
  
   # interval example, 间隔执行, 每30秒执行一次
   @scheduler.task('interval', id='test_job_1', seconds=10, misfire_grace_time=900)
   def test_job1():
     print('Job 1 executed')
  #
  #
  # cron examples, 每分钟执行一次
  # @scheduler.task('cron', id='test_job_2', second=30)
  # def test_job2():
  #     print('Job 2 executed')
  #     task_time = time.strftime("%Y-%m-%d %H:%M:%S")
  #     ll = 'echo {test1} >> time_test.txt'.format(test1=task_time)
  #     os.system(ll)
  #
  #
  # # 每周执行一次
  # @scheduler.task('cron', id='test_job_3', week='*', day_of_week='sun')
  # def test_job3():
  #     print('Job 3 executed')