Flask-apscheduler 使用

examples

1. 最简单的使用例子

from flask import Flask
from flask_apscheduler import APScheduler


# set configuration values
class Config():
    SCHEDULER_API_ENABLE = True


# create app
app = Flask(__name__)
app.config.from_object(Config)

# initialize scheduler
scheduler = APScheduler()
# if you dont want use config , you can set options here
# scheduler.api_enabled = True
scheduler.init_app(app)


# jobs can be added to the scheduler when the app starts, and should be imported before app.run() is called'
@scheduler.task('interval', id='do_job_1', seconds=10, misfire_grace_time=900)
def job1():
    print('i am job1')


scheduler.start()

if __name__ == "__main__":
    app.run()

2. 如果自己想配置config，并且不想使用装饰器

# using other configration
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore  # 用sqlalchemy在sql中储存jobs
from flask import Flask

from flask_apscheduler import APScheduler


class Config():
    """app comfigration"""
    JOBS = [
        {
            "id": "job1",
            "func": "advanced:job1",  # 前面是job所处python文件的地址,:后面不能有空字符
            "args": (1, 2),
            "trigger": "interval",
            "seconds": 3,
        }
    ]

    # SCHEDULER_JOBSTORES = {"default": SQLAlchemyJobStore(url="sqlite://")}
    #
    # SCHEDULER_EXECUTORS = {"default": {"type": "threadpool", "max_workers": 20}}
    # 
    # SCHEDULER_JOB_DEFAULTS = {"coalesce": False, "max_instances": 3}

    SCHEDULER_API_ENABLED = True


def job1(var_one, var_two):
    """Demo job function.
    :param var_two:
    :param var_two:
    """
    print(str(var_one) + " " + str(var_two))


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

    scheduler = APScheduler()
    scheduler.init_app(app)
    scheduler.start()

    app.run()

3. 使用工厂
   1. 设置配置文件settings.py，用于区分生产环境和开发环境

   class Config:
       DEBUG = False
       TESTING = True
       
   class DevelopmentConfig(Config):
       DEBUG = True
   

   2. apscheduler 和其他的一些功能模块一般放在extension中

   """ initialize any app extensions """
   from flask_apscheduler import APScheduler
   
   scheduler = APScheduler()
   

   3. tasks

   from .extensions import scheduler
   
   @scheduler.task("interval", id='job_sync', seconds=10, 
                   max_instances=1, start_date='2000-01-01 12:19:00')
   def task1():
       """
       Sample task1
       added when app starts
       :return: 
       """
       print('running task 1')
       with scheduler.app.app_context():
           print(scheduler.app.config)
   
   
   def task2():
       """
       added when /add url is visted
       :return: 
       """
       print("running task 2")
   

   4. init.py 统领

   import logging
   import os
   
   from flask import Flask
   from .settings import DevelopmentConfig
   from .extensions import scheduler
   
   
   def create_app():
       """create a new app instance"""
   
       def is_debug_mode():
           debug = os.environ.get("FLASK_DEBUG")
           if not debug:
               return os.environ.get("FLASK_ENV") == 'development'
           return debug.lower() not in ("0", "flase", "no")
   
       def is_werkzeug_reloader_process():
           """Get werkzeug status."""
           return os.environ.get("WERKZEUG_RUN_MAIN") == "true"
   
       app = Flask(__name__)
   
       app.config.from_object(DevelopmentConfig)
       scheduler.init_app(app)
   
       logging.getLogger("apscheduler").setLevel(logging.INFO)
   
       with app.app_context():
           if is_debug_mode() and not is_werkzeug_reloader_process():
               pass
           else:
               from . import task
   
               scheduler.start()
           from . import event, web
           app.register_blueprint(web.web_bp)
           return app
   

   5. 然后创建一个入口文件就可以了

   from factory import create_app
   
   if __name__ == "__main__":
       app = create_app()
       app.run()
   

注释

@scheduler.task("interval", id='job_sync', seconds=10, 
                max_instances=1, start_date='2000-01-01 12:19:00')

1. interval是trigger，触发器，触发器有三种
   date：use when you want to run the job just once at a certain point of time
   interval：use when you want to run the job at fixed intervals of time
   cron：use when you want to run the job periodically at certain time(s) of day

   Fields greater than the least significant explicitly defined field default to * while lesser fields default to their minimum values except for week and day_of_week which default to *. For example, day=1, minute=20 is equivalent to year=’*’, month=’*’, day=1, week=’*’, day_of_week=’*’, hour=’*’, minute=20, second=0. The job will then execute on the first day of every month on every year at 20 minutes of every hour. The code examples below should further illustrate this behavior.

   对于时间设定的详细解说： https://apscheduler.readthedocs.io/en/stable/modules/triggers/cron.html#module-apscheduler.triggers.cron