关于flask:FlaskAPScheduler详细配置使用附带API调用

谁再不提前告诉我，就间接复制我的文章而后发表，全家生孩子没屁眼。

原文链接：https://segmentfault.com/a/11…

作者：SyntaxError

1. 配置

from App.tasks.DatabaseTask import send_ding_test  # 我的工作函数
from flask_apscheduler.auth import HTTPBasicAuth
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore


class Config(object):

    JOBS = [
        # interval 定时执行（从 start_date 到 end_date，距离 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': True  # 从新执行程序时，会将 jobStore 中的工作替换掉
        },
        # date 一次执行
        {
            'id': 'job1',
            'func': send_ding_test,
            'trigger': 'date',
            'run_date': '2021-01-30 11:22:00',
            'replace_existing': True
        },
        # cron 式定时调度，相似 linux 的 crontab
        {
            'id': 'job3',
            'func': send_ding_test,
            'trigger': 'cron',
            'day_of_week': '0-6',
            'month': '*',
            'hour': '6',
            'minute': '0',
            'second': '0',
            'replace_existing': True
        }

    ]

    # 存储定时工作（默认是存储在内存中）SCHEDULER_JOBSTORES  = {'default':SQLAlchemyJobStore(url='mysql+pymysql://xxx/xx')}
    # 设置时区，时区不统一会导致定时工作的工夫谬误
    SCHEDULER_TIMEZONE = 'Asia/Shanghai'
    # 肯定要开启 API 性能，这样才能够用 api 的形式去查看和批改定时工作
    SCHEDULER_API_ENABLED = True
    # api 前缀（默认是 /scheduler）SCHEDULER_API_PREFIX = '/scheduler'
    # 配置容许执行定时工作的主机名
    SCHEDULER_ALLOWED_HOSTS = ['*']
    # auth 验证。默认是敞开的，SCHEDULER_AUTH = HTTPBasicAuth()
    # 设置定时工作的执行器（默认是最大执行数量为 10 的线程池）SCHEDULER_EXECUTORS = {'default': {'type': 'threadpool', 'max_workers': 10}}
    # 另外 flask-apscheduler 内有日志记录器。name 为 apscheduler.scheduler 和 apscheduler.executors.default。如果须要保留日志，则须要对此日志记录器进行配置
    

2. 初始化 app

import logging
from flask import Flask
from config import Config  # 上边的配置文件
from flask_apscheduler.scheduler import APScheduler
from logging.handlers import RotatingFileHandler


# flask-apscheduler 内置有日志器，为了让外部的日志器打印的内容输入，我这里做了个配置
# 创立日志记录器，指明日志保留门路，每个日志大小，保留日志文件个数下限
file_log_handler = RotatingFileHandler('logs/runserver.log', maxBytes=1024*1024*100, backupCount=5)
# 创立日志的记录格局,]，日志等级，记录时间，报错地位，行数，日志信息
formatter = logging.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)
# 为全局日志对象增加日志记录器
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()
    
    # 配置 api 权限验证的回调函数
    @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.API 调用

Flask-APScheduler 内置了丰盛的 api 接口，能够让开发者动静的查看和更改定时工作，十分不便。这里我找到外部的一部分源码，能够看到所有的 api 接口的调用形式。

    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')

 
调用形式我这里用 postman 演示。首先是查看定时工作状态：

 
查看所有的定时工作：

 
立即执行一次定时工作：

 
还有就是，这些 api 能够带权限验证的，还记得我在配置那里配置了一项参数是：

SCHEDULER_AUTH = HTTPBasicAuth()

减少这项配置后，所有的 api 调用都必须携带 AUTHORIZATION。我这里说下怎么应用。在 init_app 那里我曾经设置了用户名明码（guest，guest），在申请接口时，须要携带加密后的用户名明码，加密形式如下：

import base64

# 用户名明码用: 分隔
string = "guest:guest".encode('utf-8')

a = base64.b64encode(string)
# print(a)
# b'Z3Vlc3Q6Z3Vlc3Q='
print(b"basic" + a)  # 最初必须拼接上 basic
# b'basic Z3Vlc3Q6Z3Vlc3Q='

而后将这个字符串放在申请头中，申请头的键就是 AUTHORIZATION：

大家有什么问题，能够在文章下方留言，我看到后会立刻回复
其余很常见的问题，比方定时反复执行的问题，网上曾经有很多解决方案了，这里不再论述