Flask-Apscheduler detailed configuration usage (with API calls)

If anyone does not inform me in advance, they will copy my article and publish it directly. The whole family will have children without assholes.

The original link: https://segmentfault.com/a/11…

Author: SyntaxError

1. The configuration

From app.tasks. DatabaseTask import send_ding_test # My task function from flask_apscheduler. Auth import HttpBasicauth from Flask_apscheduler apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore class Config(object): JOBS = [# interval] (from start_date to end_date, interval 20s) # func 'App.tasks.DatabaseTask:send_ding_test' { 'id': 'job2', 'func': send_ding_test, 'trigger': 'interval', 'start_date': '2021-01-27 13:31:00', 'end_date': '2021-01-27 13:33:00', 'seconds': 20, 'replace_existing': {'id': 'job1', 'func': send_ding_test, 'trigger': 'date', 'run_date': '2021-01-30 11:22:00', 'replace_existing': True}, # crontab {'id': 'job3', 'func': send_ding_test, 'trigger': 'cron', 'day_of_week': '0-6', 'month': '*', 'hour': '6', 'minute': '0', 'second': '0', 'replace_existing': Scheduler_JobStores = {'default':SQLAlchemyJobStore(url=' MySQL + PyMySQL :// XXX /xx')} Scheduler_timezone = 'Asia/Shanghai' # Make sure the API is enabled. Scheduler_API_ENabled = True # Scheduler_API_ENabled = '/scheduler' # Scheduler_API_ENabled = '/scheduler' # Set up the hostname SCHEDULER_ALLOWED_HOSTS = ['*'] # auth validation that allows timed tasks to be performed. Scheduler_Executors = {'default': {'type': Scheduler_Executors = {'default': {'type': 'threadpool', 'max_workers': 10}} # Flask-apscheduler contains logger The name for the apscheduler. Scheduler and apscheduler. Executors. Default. If you want to save logs, you need to configure this logger

2. Initialize the app

Scheduler import logging from flask import flask from config import config # scheduler import logging from flask_apscheduler Handlers from Logging. Handlers import RotatingFileHandler # flask-apscheduler = "Handlers from Logging"; I made a configuration here # create log logger, specify log save path, each log size, File_log_handler = RotatingFileHandler('logs/runserver.log', maxBytes=1024*1024*100, BackupCount = 1) # create log format,], log level, record time, error position, line number, Formatter(FMT ="%(asctime)s - %(levelName)s - %(threadName)s:%(thread)s - %(filename)s - %(funcName)s - %(message)s", Datefmt ='%Y-%m-%d %H:% m :%S %a') file_log_handler.setFormatter(formatter) # add logger to global logger =  logging.getLogger("apscheduler") logger.addHandler(file_log_handler) logger.setLevel(logging.INFO) def create_app(): app = Flask(__name__) app.config.from_object(Config) scheduler = APScheduler() scheduler.init_app(app) scheduler.start() @scheduler.authenticate def authenticate(auth): @scheduler.authenticate def authenticate(auth): Return auth['username'] == 'guest' and auth['password'] == 'guest' app = create_app() app.run(host='127.0.0.1'); port='7788')

3. The API calls

Flask-Apscheduler has a rich built-in API interface, which allows developers to dynamically view and change the scheduled tasks, which is very convenient. Here I find the internal part of the source code, you can see all the API interface calls.

def _load_api(self): Self._add_url_route(' Get_Scheduler_Info ', '', ') Add_url_route ('add_job', '/jobs', api.add_job, '/jobs') 'POST') # for self task. _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', Self._add_url_route('run_job', '/jobs/<job_id>/run', api.run_job, 'POST')

I’m going to use PostMan to show you how to do that. First, check the status of the timed task:

View all timed tasks:

Perform a timed task immediately:

Also, these APIs can be authenticated with permissions. Remember that I configured a parameter in the configuration that was:

SCHEDULER_AUTH = HTTPBasicAuth()

With this configuration added, all API calls must carry AUTHORIZATION. Let me show you how to use it. In init_app, I have set the user name and password (guest, guest). When requesting the interface, I need to carry the encrypted user name and password. The encryption method is as follows:

String = "guest:guest". Encode ('utf-8') a = base64.b64encode(string) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print(A) # print B 'Z3Vlc3Q6Z3Vlc3Q=' print(b" lc3q6z3vlc3q =')

This string is then placed in the request header, and the key of the request header is AUTHORIZATION:

If you have any questions, you can leave a message at the bottom of the article. After I see it, I will immediately reply to other very common questions, such as the problem of timed and repeated execution. There are already many solutions on the Internet, so I will not elaborate here