Hardware and software Environment

  • windows 10 64bit
  • Anaconda3 with python 3.7
  • Pycharm 2020.1.2
  • Flask 1.1.2
  • Flask – apscheduler 1.11.0

flask-apscheduler

preface

Flask-apscheduler migrates apscheduler into flask applications, making it easy to use scheduled tasks in flask. In addition, it has the following features

  • According to theFlaskConfiguration loads the scheduler configuration
  • According to theFlaskConfigure the load task scheduler
  • Allows a specified server to run tasks
  • provideRESTful APIAdministrative tasks, that is, remote administrative tasks
  • forRESTful APIProvide the certification

Use the sample

Before using it, we need to install this module, using PIP

pip install flask-apscheduler
Copy the code

Flask-apscheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler: Flask-apScheduler

from flask import Flask
from flask_apscheduler import 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)


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

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

    app.run()
Copy the code

In the Config class, there is a list of JOBS, and each element is a task. In the example code above, there is only one task, an interval task, which is executed every 3 seconds. The specific task method is Add, which receives 2 parameters. The format of the value after func is module name: method name

Flask-apscheduler initialization is necessary after the flask app is instantiated and before it is run

scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
Copy the code

By performing the above project, we can get

(FlaskTutorial) D:\xugaoxiang\FlaskTutorial\Flask-20-apscheduler>python run.py
 * Serving Flask app "run" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on http://127.0. 01.:5000/ (Press CTRL+C to quit)
3
3
3
Copy the code

Can we set the above SCHEDULER_API_ENABLED = True, by visiting http://127.0.0.1:5000/scheduler, the scheduler is the default prefix RESTful API

By looking at the source file scheduler.py, we can see what flask-apScheduler provides us with RESTful apis

    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')
Copy the code

Method is to use them, I have to check all the tasks in the current application, for example, can be accessed using the GET method to http://127.0.0.1:5000/scheduler/jobs

If you’re not comfortable with configuration, you can also use decorators, as shown in the following example

from flask import Flask
from flask_apscheduler import APScheduler


class Config(object) :
    SCHEDULER_API_ENABLED = True


scheduler = APScheduler()


@scheduler.task('interval'.id='do_job_1', seconds=30)
def job1() :
    print('Job 1 executed')


# cron examples
@scheduler.task('cron'.id='do_job_2', minute=The '*')
def job2() :
    print('Job 2 executed')


@scheduler.task('cron'.id='do_job_3', week=The '*', day_of_week='sun')
def job3() :
    print('Job 3 executed')


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

    scheduler.init_app(app)
    scheduler.start()

    app.run()
Copy the code

Finally, let’s look at a slightly more complicated configuration

class Config(object) :
    JOBS = [
        {
            'id': 'job1'.'func': 'run:job1'.'args': (1.2),
            'trigger': 'interval'.'seconds': 10
        }
    ]

    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
Copy the code

SCHEDULER_JOBSTORES refers to the job store, which we store in SQLite. SCHEDULER_EXECUTORS refer to the configuration of the actuator, use the threadpool threadpool, and set the maximum number of threads to 20. SCHEDULER_JOB_DEFAULTS are some configurations of tasks, where

  • Coalesce refers to a situation in which a task is accumulated for several times and fails to run. For example, the system recovers after being suspended for two minutes. For example, a supervisor process monitoring tool runs a task once every minute, which is supposed to run twice within two minutes but fails to run. If coalesce is True, the coalesce task is executed only once. If coalesce is False, the coalesce task is executed twice

  • Max_instance means that at most several instances of a task are running at the same time

SCHEDULER_API_ENABLED 'specifies whether the API is enabledCopy the code

In addition to the usual configuration above, there are a few more

SCHEDULER_TIMEZONE          # Configure time zone
SCHEDULER_API_PREFIX        Configure the API route prefix
SCHEDULER_ENDPOINT_PREFIX   Configure the API route suffix
SCHEDULER_ALLOWED_HOSTS     Configure access whitelist
SCHEDULER_AUTH              # Configure the authentication center
Copy the code

When deploying the Flask application on the server, you often need to set the time zone; otherwise, an error will be reported

class Config(object) :
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'
Copy the code

Configure the RESTful API route prefix and suffix

class Config(object) :
    SCHEDULER_API_PREFIX ='/waws'
Copy the code

Then access all task routing will be from the original http://127.0.0.1:5000/scheduler/jobs to http://127.0.0.1:5000/waws/jobs

If you want to allow only certain hosts to access, you can set the whitelist. If you want to allow all hosts to access, you can also write *

class Config(object):
    SCHEDULER_ALLOWED_HOSTS = ['waws.com']
Copy the code

If you want to add authentication, you can do this

from flask import Flask
from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth


class Config(object) :
    JOBS = [
        {
            'id': 'job1'.'func': 'run2:add'.'args': (1.2),
            'trigger': 'interval'.'seconds': 3
        }
    ]

    SCHEDULER_API_ENABLED = True
    SCHEDULER_AUTH = HTTPBasicAuth()


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


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

    scheduler = APScheduler()
    # it is also possible to set the authentication directly
    # scheduler.auth = HTTPBasicAuth()
    scheduler.init_app(app)
    scheduler.start()

    @scheduler.authenticate
    def authenticate(auth) :
        return auth['username'] = ='guest' and auth['password'] = ='guest'

    app.run()
Copy the code

Before visiting the scheduler, authentication is performed first. If the conditions are met, such as username=guest and password=guest, which are passed through the HTTP request above, the access can be continued; otherwise, the access is rejected.