Perform scheduled tasks in the Flask

This is the 24th day of my participation in the August Text Challenge.More challenges in August

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

preface

The basic use of Apscheduler has been described in the Python utility module Apscheduler. 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.0.1: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='*')
def job2():
    print('Job 2 executed')


@scheduler.task('cron', id='do_job_3', week='*', 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 enabled

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

SCHEDULER_TIMEZONE # Configure time zone SCHEDULER_API_PREFIX # Configure API route prefix SCHEDULER_ENDPOINT_PREFIX # Configure API route suffix SCHEDULER_ALLOWED_HOSTS SCHEDULER_AUTH # Configure the authentication centerCopy 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 ='/xugaoxiang'
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/xugaoxiang/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 = ['xugaoxiang.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.

Download the source code

Github.com/xugaoxiang/…

Flask Series of Tutorials

For more Flask tutorials, go ahead

Xugaoxiang.com/category/py…

The resources

Github.com/viniciuschi…
The utility module apscheduler

Perform scheduled tasks in the Flask

Hardware and software Environment

preface

Use the sample

Download the source code

Flask Series of Tutorials

The resources

Related Posts

Unable to execute jAR-file: “No master list properties” problem handling

Lambda expressions and functional interfaces

Java thread creation in three ways and comparison