乐趣区

关于flask:flask-FlaskAPScheduler-配置计划任务

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')
    
    
    
退出移动版